This is an automated email from the ASF dual-hosted git repository. gyfora pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2b20571445d7d8d3e5695a61ac9f93a1da64dcc1 Author: Gyula Fora <[email protected]> AuthorDate: Sun Jul 19 21:49:02 2026 +0200 [FLINK-40177][state-processor-api] Flattened keyed state table mapping and implementation Adds the flattened keyed-state table (selected via STATE_READER_MODE.KEYED_FLAT): a single named LIST/MAP state is exposed as (key, list_index, list_value) or (key, map_key, map_value) rows instead of one column per state. - FlattenedStateTableMapping resolves and validates the fixed 3-column schema against the target state's actual serializers. - SingleColumnStateMapping is the shared interface for "flattened", single value-column table mappings (implemented by FlattenedStateTableMapping now, and reused by the windowed-flattened mapping added in a later commit). - AbstractSingleColumnScanProvider/FlattenedSavepointDynamicTableSource provide the shared scan-provider and DynamicTableSource machinery parameterized over a SingleColumnStateMapping, mirroring AbstractMultiColumnScanProvider/ AbstractSavepointDynamicTableSource from the previous commit. - FlattenedKeyedStateReader reads the target state via the DataStream state processor API and flattens each key's list/map entries into rows. - SavepointDynamicTableSourceFactory wires STATE_READER_MODE.KEYED_FLAT to this machinery, and KeyedTableMappingSupport gains the flattened-schema inference/validation and serializer-resolution helpers shared with the windowed-flattened mapping. --- .../apache/flink/state/api/StateTableUtils.java | 140 ++++++++++++ .../apache/flink/state/catalog/StateCatalog.java | 60 +++++- .../table/AbstractSingleColumnScanProvider.java | 73 +++++++ .../state/table/FlattenedKeyedStateReader.java | 148 +++++++++++++ .../FlattenedSavepointDataStreamScanProvider.java | 56 +++++ .../FlattenedSavepointDynamicTableSource.java | 76 +++++++ .../state/table/FlattenedStateTableMapping.java | 234 +++++++++++++++++++++ .../table/SavepointDynamicTableSourceFactory.java | 73 +++++++ .../state/table/SingleColumnStateMapping.java | 47 +++++ .../flink/state/table/TableMappingSupport.java | 170 ++++++++++++++- .../state/catalog/StateCatalogDiscoveryITCase.java | 10 +- 11 files changed, 1074 insertions(+), 13 deletions(-) diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java index 7b681c88a8a..b6a4eacba30 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java @@ -42,7 +42,10 @@ import org.apache.flink.table.api.Schema; import org.apache.flink.table.catalog.CatalogTable; import org.apache.flink.table.factories.FactoryUtil; import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.MapType; import org.apache.flink.table.types.logical.VarBinaryType; import org.apache.flink.table.types.utils.LogicalTypeDataTypeConverter; @@ -262,11 +265,148 @@ public final class StateTableUtils { Schema schema = schemaBuilder.build(); Map<String, String> options = buildBaseConnectorOptions(statePath, operatorIdentifier); + options.put( + SavepointConnectorOptions.STATE_READER_MODE.key(), + (windowType == null + ? SavepointConnectorOptions.StateReaderMode.KEYED + : SavepointConnectorOptions.StateReaderMode.WINDOWED) + .toString()); withStateBackendType(options, metadata, operatorIdentifier); return CatalogTable.newBuilder().schema(schema).options(options).build(); } + /** + * Builds a {@link CatalogTable} exposing a single keyed LIST or MAP state flattened into one + * row per list element / map entry, rather than one row per key. + * + * <p>The resulting table has 3 columns, with a composite primary key on {@code state_key} and + * the sub-key column (the {@code state_key} value repeats across rows belonging to the same + * key, but the pair uniquely identifies a row). The third column has a fixed name — not the + * state's own name, to avoid collisions with other (reserved) column names: + * + * <ul> + * <li>LIST: {@code (state_key, list_index, list_value)}, primary key {@code (state_key, + * list_index)} + * <li>MAP: {@code (state_key, map_key, map_value)}, primary key {@code (state_key, map_key)} + * </ul> + * + * @param metadata the checkpoint metadata the operator belongs to + * @param schemaInfo the schema information returned by {@link #getKeyedStateSchema} + * @param stateName the name of the LIST or MAP state to flatten + * @param statePath the path to the savepoint / checkpoint + * @param operatorIdentifier identifies the operator whose state to read + * @return a {@link CatalogTable} ready for registration + */ + public static CatalogTable getFlattenedStateCatalogTable( + CheckpointMetadata metadata, + KeyedStateSchemaInfo schemaInfo, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier) { + return buildFlattenedKeyedCatalogTable( + metadata, schemaInfo, stateName, statePath, operatorIdentifier, false); + } + + /** + * Builds a {@link CatalogTable} exposing a single LIST or MAP state flattened into one row per + * list element / map entry, either plain-keyed ({@code windowed == false}, see {@link + * #getFlattenedStateCatalogTable}) or namespaced ({@code windowed == true}). + */ + private static CatalogTable buildFlattenedKeyedCatalogTable( + CheckpointMetadata metadata, + KeyedStateSchemaInfo schemaInfo, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier, + boolean windowed) { + + KeyedStateSchemaInfo.StateEntryInfo entryInfo = schemaInfo.stateSchemas.get(stateName); + if (entryInfo == null) { + throw new IllegalArgumentException( + "State '" + + stateName + + "' not found for operator '" + + operatorIdentifier + + "'."); + } + if (entryInfo.stateType != SavepointConnectorOptions.StateType.LIST + && entryInfo.stateType != SavepointConnectorOptions.StateType.MAP) { + throw new IllegalArgumentException( + "Flattened state tables are only supported for LIST and MAP states, but '" + + stateName + + "' is " + + entryInfo.stateType + + "."); + } + if (windowed && entryInfo.windowLogicalType == null) { + throw new IllegalArgumentException( + "State '" + + stateName + + "' is not a namespaced state for operator '" + + operatorIdentifier + + "'."); + } + + Schema.Builder schemaBuilder = Schema.newBuilder(); + schemaBuilder.column( + "state_key", LogicalTypeDataTypeConverter.toDataType(schemaInfo.keyType).notNull()); + if (windowed) { + schemaBuilder.column( + "state_window", + LogicalTypeDataTypeConverter.toDataType(entryInfo.windowLogicalType).notNull()); + } + + String subKeyColumnName = addFlattenedValueColumns(schemaBuilder, entryInfo); + if (!windowed) { + schemaBuilder.primaryKeyNamed( + "PK_state_key_" + subKeyColumnName, "state_key", subKeyColumnName); + } + Schema schema = schemaBuilder.build(); + + Map<String, String> options = buildBaseConnectorOptions(statePath, operatorIdentifier); + options.put( + SavepointConnectorOptions.STATE_READER_MODE.key(), + (windowed + ? SavepointConnectorOptions.StateReaderMode.WINDOWED_FLAT + : SavepointConnectorOptions.StateReaderMode.KEYED_FLAT) + .toString()); + options.put(SavepointConnectorOptions.FLATTENED_STATE_NAME.key(), stateName); + withStateBackendType(options, metadata, operatorIdentifier); + + return CatalogTable.newBuilder().schema(schema).options(options).build(); + } + + /** + * Adds the LIST- or MAP-shaped sub-key and value columns (e.g. {@code (list_index, list_value)} + * or {@code (map_key, map_value)}) for a flattened state table, and returns the sub-key + * column's name. + */ + private static String addFlattenedValueColumns( + Schema.Builder schemaBuilder, KeyedStateSchemaInfo.StateEntryInfo entryInfo) { + LogicalType valueType; + String subKeyColumnName; + String valueColumnName; + if (entryInfo.stateType == SavepointConnectorOptions.StateType.LIST) { + valueType = ((ArrayType) entryInfo.logicalType).getElementType(); + subKeyColumnName = "list_index"; + valueColumnName = "list_value"; + schemaBuilder.column( + subKeyColumnName, + LogicalTypeDataTypeConverter.toDataType(new BigIntType(false))); + } else { + MapType mapType = (MapType) entryInfo.logicalType; + valueType = mapType.getValueType(); + subKeyColumnName = "map_key"; + valueColumnName = "map_value"; + schemaBuilder.column( + subKeyColumnName, + LogicalTypeDataTypeConverter.toDataType(mapType.getKeyType()).notNull()); + } + schemaBuilder.column(valueColumnName, LogicalTypeDataTypeConverter.toDataType(valueType)); + return subKeyColumnName; + } + // ------------------------------------------------------------------------- // Private helpers // ------------------------------------------------------------------------- diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java index d55b10cdcdd..66c6b174859 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java @@ -61,9 +61,11 @@ import javax.annotation.Nullable; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; /** * A read-only Flink SQL catalog that discovers checkpoints and savepoints from a configured set of @@ -108,6 +110,7 @@ public class StateCatalog extends AbstractCatalog { public static final String OPERATOR_UID_PREFIX = "uid_"; public static final String OPERATOR_ID_PREFIX = "id_"; public static final String OPERATOR_TABLE_SUFFIX = "_keyed"; + public static final String FLAT_STATE_TABLE_SUFFIX = "_keyed_flat"; private static final CatalogDatabase EMPTY_DATABASE = new CatalogDatabaseImpl(Collections.emptyMap(), ""); @@ -214,15 +217,33 @@ public class StateCatalog extends AbstractCatalog { throw new DatabaseNotExistException(getName(), databaseName); } List<String> tables = new ArrayList<>(); + tables.add(METADATA_TABLE); + Set<String> seen = new LinkedHashSet<>(); try { CheckpointMetadata metadata = SavepointLoader.loadSavepointMetadata(snapshotPath.get()); for (OperatorIdentifier opId : StateTableUtils.getOperatorIdentifiers(metadata)) { for (ResolvedTable candidate : candidateTablesForOperator(metadata, opId)) { - tables.add( + String name = tableName( candidate.operatorIdentifier, candidate.kind, - candidate.stateName)); + candidate.stateName); + // Two distinct (operator, state) combinations can legitimately derive the + // same table name, since operator UIDs and state names may themselves + // contain underscores (see #tableName). Skip and warn rather than exposing + // the same name twice, mirroring how SnapshotDiscovery#list handles + // colliding database names. + if (!seen.add(name)) { + LOG.warn( + "Table name '{}' is ambiguous between multiple operators/states " + + "in database '{}' and only the first one found is " + + "exposed. Consider renaming the colliding operator " + + "UID(s) or state name(s).", + name, + databaseName); + continue; + } + tables.add(name); } } } catch (IOException e) { @@ -269,6 +290,18 @@ public class StateCatalog extends AbstractCatalog { snapshotPath.get(), resolved.operatorIdentifier); } + case KEYED_FLAT: + { + KeyedStateSchemaInfo schemaInfo = + StateTableUtils.getKeyedStateSchema( + metadata, resolved.operatorIdentifier); + return StateTableUtils.getFlattenedStateCatalogTable( + metadata, + schemaInfo, + resolved.stateName, + snapshotPath.get(), + resolved.operatorIdentifier); + } default: throw new IllegalStateException("Unhandled table kind " + resolved.kind); } @@ -296,6 +329,10 @@ public class StateCatalog extends AbstractCatalog { CheckpointMetadata metadata = SavepointLoader.loadSavepointMetadata(snapshotPath.get()); return resolveTable(metadata, tableName).isPresent(); } catch (IOException e) { + LOG.warn( + "Failed to load checkpoint metadata while checking existence of table '{}'", + tablePath, + e); return false; } } @@ -560,7 +597,7 @@ public class StateCatalog extends AbstractCatalog { /** * Table name for a {@code kind} of operator state, optionally scoped to one flattened/non-keyed - * state (see {@link #OPERATOR_TABLE_SUFFIX}). + * state (see {@link #OPERATOR_TABLE_SUFFIX}/{@link #FLAT_STATE_TABLE_SUFFIX}). * * <p>{@code stateName} must be {@code null} for {@link StateReaderMode#KEYED}/{@link * StateReaderMode#WINDOWED} (the general keyed/namespaced table, one per operator) and non-null @@ -569,7 +606,9 @@ public class StateCatalog extends AbstractCatalog { * within an operator). */ private static final Map<StateReaderMode, String> TABLE_SUFFIXES = - Map.of(StateReaderMode.KEYED, OPERATOR_TABLE_SUFFIX); + Map.of( + StateReaderMode.KEYED, OPERATOR_TABLE_SUFFIX, + StateReaderMode.KEYED_FLAT, FLAT_STATE_TABLE_SUFFIX); static String tableName( OperatorIdentifier opId, StateReaderMode kind, @Nullable String stateName) { @@ -620,6 +659,7 @@ public class StateCatalog extends AbstractCatalog { try { candidates = candidateTablesForOperator(metadata, opId); } catch (IOException e) { + LOG.warn("Failed to load state schema for operator '{}'. Skipping.", opId, e); continue; } for (ResolvedTable candidate : candidates) { @@ -633,8 +673,8 @@ public class StateCatalog extends AbstractCatalog { } /** - * Enumerates every table that {@code opId} contributes: currently just the general keyed table - * (if any plain per-key state is registered). + * Enumerates every table that {@code opId} contributes: the general keyed table (if any plain + * per-key state is registered), plus one flattened table per LIST/MAP keyed state. * * <p>Shared by {@link #listTables} (which collects names for every candidate) and {@link * #resolveTable} (which matches candidates against a target name), so that adding a new state @@ -647,6 +687,14 @@ public class StateCatalog extends AbstractCatalog { KeyedStateSchemaInfo schemaInfo = StateTableUtils.getKeyedStateSchema(metadata, opId); if (!schemaInfo.stateSchemas.isEmpty()) { candidates.add(new ResolvedTable(opId, StateReaderMode.KEYED)); + for (Map.Entry<String, KeyedStateSchemaInfo.StateEntryInfo> entry : + schemaInfo.stateSchemas.entrySet()) { + StateType stateType = entry.getValue().stateType; + if (stateType == StateType.LIST || stateType == StateType.MAP) { + candidates.add( + new ResolvedTable(opId, StateReaderMode.KEYED_FLAT, entry.getKey())); + } + } } return candidates; diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractSingleColumnScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractSingleColumnScanProvider.java new file mode 100644 index 00000000000..590324b206f --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractSingleColumnScanProvider.java @@ -0,0 +1,73 @@ +/* + * 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.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.filter.SavepointKeyFilter; +import org.apache.flink.state.api.schema.StateSchemaInfo; +import org.apache.flink.table.types.logical.RowType; + +import javax.annotation.Nullable; + +import java.util.Map; +import java.util.function.Supplier; + +/** + * Base for scan providers whose mapping describes exactly one flattened LIST/MAP state via a single + * fixed descriptor (see {@link SingleColumnStateMapping}): {@link + * FlattenedSavepointDataStreamScanProvider} and {@link + * WindowFlattenedSavepointDataStreamScanProvider}. + */ +@Internal +abstract class AbstractSingleColumnScanProvider<M extends SingleColumnStateMapping> + extends AbstractSavepointDataStreamScanProvider<M> { + + protected AbstractSingleColumnScanProvider( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier<M> mappingSupplier, + final RowType rowType, + @Nullable final SavepointKeyFilter keyFilter) { + super(stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType, keyFilter); + } + + @Override + @SuppressWarnings("rawtypes") + protected final void prepareStateDescriptors(M mapping) { + Map<String, StateSchemaInfo> fallbackSchemas = + loadFallbackSchemas( + isSerializerMissing( + mapping.getStateType(), + mapping.getMapKeyTypeSerializer(), + mapping.getValueTypeSerializer())); + + StateDescriptor<?, ?> descriptor = + buildStateDescriptor( + mapping.getStateName(), + mapping.getStateType(), + StateDescriptor.Type.UNKNOWN, + mapping.getMapKeyTypeSerializer(), + mapping.getValueTypeSerializer(), + fallbackSchemas); + mapping.setStateDescriptor(descriptor); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedKeyedStateReader.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedKeyedStateReader.java new file mode 100644 index 00000000000..11592e48502 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedKeyedStateReader.java @@ -0,0 +1,148 @@ +/* + * 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.state.table; + +import org.apache.flink.api.common.functions.OpenContext; +import org.apache.flink.api.common.state.ListState; +import org.apache.flink.api.common.state.ListStateDescriptor; +import org.apache.flink.api.common.state.MapState; +import org.apache.flink.api.common.state.MapStateDescriptor; +import org.apache.flink.api.common.state.State; +import org.apache.flink.state.api.functions.KeyedStateReaderFunction; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; +import org.apache.flink.util.Collector; + +import java.util.Map; + +/** + * Reads a single flattened keyed list/map state, emitting one row per list element / map entry + * instead of one row per key: {@code (state_key, index, value)} for LIST, {@code (state_key, + * map_key, value)} for MAP. + * + * <p>Shares value-conversion logic ({@link StateValueConverter}) with {@link KeyedStateReader}. + */ +@SuppressWarnings({"rawtypes", "unchecked"}) +public class FlattenedKeyedStateReader extends KeyedStateReaderFunction<Object, RowData> { + + private final RowType rowType; + private final FlattenedStateTableMapping mapping; + private final StateValueConverter converter = new StateValueConverter(); + + private transient State state; + + public FlattenedKeyedStateReader(RowType rowType, FlattenedStateTableMapping mapping) { + this.rowType = rowType; + this.mapping = mapping; + } + + @Override + public void open(OpenContext openContext) throws Exception { + switch (mapping.getStateType()) { + case LIST: + state = + getRuntimeContext() + .getListState((ListStateDescriptor) mapping.getStateDescriptor()); + break; + + case MAP: + state = + getRuntimeContext() + .getMapState((MapStateDescriptor) mapping.getStateDescriptor()); + break; + + default: + throw new UnsupportedOperationException( + "Unsupported flattened state type: " + mapping.getStateType()); + } + } + + @Override + public void close() { + state = null; + } + + @Override + public void readKey(Object key, Context context, Collector<RowData> out) throws Exception { + LogicalType keyLogicalType = + rowType.getFields() + .get(FlattenedStateTableMapping.STATE_KEY_COLUMN_INDEX) + .getType(); + Object convertedKey = converter.getValue(keyLogicalType, key); + + switch (mapping.getStateType()) { + case LIST: + readList(convertedKey, out); + break; + + case MAP: + readMap(convertedKey, out); + break; + + default: + throw new UnsupportedOperationException( + "Unsupported flattened state type: " + mapping.getStateType()); + } + } + + private void readList(Object convertedKey, Collector<RowData> out) throws Exception { + LogicalType valueLogicalType = + rowType.getFields().get(FlattenedStateTableMapping.VALUE_COLUMN_INDEX).getType(); + LogicalType indexLogicalType = + rowType.getFields().get(FlattenedStateTableMapping.SUB_KEY_COLUMN_INDEX).getType(); + + Iterable<Object> values = (Iterable<Object>) ((ListState) state).get(); + converter.writeListRows( + values, + () -> { + GenericRowData row = new GenericRowData(RowKind.INSERT, 3); + row.setField(FlattenedStateTableMapping.STATE_KEY_COLUMN_INDEX, convertedKey); + return row; + }, + FlattenedStateTableMapping.SUB_KEY_COLUMN_INDEX, + FlattenedStateTableMapping.VALUE_COLUMN_INDEX, + indexLogicalType, + valueLogicalType, + out); + } + + private void readMap(Object convertedKey, Collector<RowData> out) throws Exception { + LogicalType valueLogicalType = + rowType.getFields().get(FlattenedStateTableMapping.VALUE_COLUMN_INDEX).getType(); + LogicalType mapKeyLogicalType = + rowType.getFields().get(FlattenedStateTableMapping.SUB_KEY_COLUMN_INDEX).getType(); + + Iterable<Map.Entry<Object, Object>> entries = ((MapState) state).entries(); + converter.writeMapRows( + entries, + () -> { + GenericRowData row = new GenericRowData(RowKind.INSERT, 3); + row.setField(FlattenedStateTableMapping.STATE_KEY_COLUMN_INDEX, convertedKey); + return row; + }, + FlattenedStateTableMapping.SUB_KEY_COLUMN_INDEX, + FlattenedStateTableMapping.VALUE_COLUMN_INDEX, + mapKeyLogicalType, + valueLogicalType, + out); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedSavepointDataStreamScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedSavepointDataStreamScanProvider.java new file mode 100644 index 00000000000..650f64d93eb --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedSavepointDataStreamScanProvider.java @@ -0,0 +1,56 @@ +/* + * 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.state.table; + +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.SavepointReader; +import org.apache.flink.state.api.filter.SavepointKeyFilter; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.RowType; + +import javax.annotation.Nullable; + +import java.util.function.Supplier; + +/** + * Savepoint data stream scan provider for a single flattened keyed LIST/MAP state, emitting one row + * per list element / map entry (see {@link FlattenedKeyedStateReader}). + */ +@SuppressWarnings("rawtypes") +public class FlattenedSavepointDataStreamScanProvider + extends AbstractSingleColumnScanProvider<FlattenedStateTableMapping> { + + public FlattenedSavepointDataStreamScanProvider( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier<FlattenedStateTableMapping> mappingSupplier, + final RowType rowType, + @Nullable final SavepointKeyFilter keyFilter) { + super(stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType, keyFilter); + } + + @Override + protected DataStream<RowData> readState( + SavepointReader savepointReader, FlattenedStateTableMapping mapping) throws Exception { + return readVoidNamespaceKeyedState( + savepointReader, mapping, new FlattenedKeyedStateReader(rowType, mapping)); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedSavepointDynamicTableSource.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedSavepointDynamicTableSource.java new file mode 100644 index 00000000000..437e29de846 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedSavepointDynamicTableSource.java @@ -0,0 +1,76 @@ +/* + * 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.state.table; + +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.filter.SavepointKeyFilter; +import org.apache.flink.table.types.logical.RowType; + +import javax.annotation.Nullable; + +import java.util.function.Supplier; + +/** + * Dynamic source for a table exposing a single flattened LIST/MAP state, i.e. every table kind + * whose mapping describes exactly one such state via a single fixed descriptor (see {@link + * SingleColumnStateMapping}): the plain keyed variant ({@link FlattenedStateTableMapping}, 3-column + * schema) and the namespaced (e.g. window-scoped) variant ({@link + * WindowFlattenedStateTableMapping}, 4-column schema). + * + * <p>Unlike {@link SavepointDynamicTableSource}, projection push-down is not supported: the schema + * is always exactly {@code (state_key[, state_window], index/map_key, value)}. Filter push-down on + * {@code state_key} is supported (via {@link SavepointKeyFilter}), pruning key groups/keys even + * though {@code state_key} is only part of the composite primary key. + */ +public class FlattenedSavepointDynamicTableSource<M extends SavepointStateMapping> + extends AbstractSavepointDynamicTableSource<M> { + + public FlattenedSavepointDynamicTableSource( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final int keyColumnIndex, + final Supplier<M> mappingSupplier, + final RowType rowType, + final String summaryString, + final ScanProviderFactory<M> scanProviderFactory) { + super( + stateBackendType, + statePath, + operatorIdentifier, + keyColumnIndex, + mappingSupplier, + rowType, + summaryString, + scanProviderFactory); + } + + @Override + protected AbstractSavepointDynamicTableSource<M> newInstance() { + return new FlattenedSavepointDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + keyColumnIndex, + mappingSupplier, + rowType, + summaryString, + scanProviderFactory); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedStateTableMapping.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedStateTableMapping.java new file mode 100644 index 00000000000..553ddeddbd4 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/FlattenedStateTableMapping.java @@ -0,0 +1,234 @@ +/* + * 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.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.UniqueConstraint; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.util.Preconditions; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.util.List; + +/** + * Maps the fixed 3-column schema of a flattened keyed list/map state table: + * + * <ul> + * <li>LIST: {@code (state_key, list_index, list_value)} + * <li>MAP: {@code (state_key, map_key, map_value)} + * </ul> + * + * <p>The third column has a fixed name ({@code list_value}/{@code map_value}) rather than being + * named after the flattened state itself, to avoid collisions with other (reserved) column names; + * the true state name is instead resolved from {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME}. + * + * <p>A flattened table always exposes exactly one keyed state and emits one row per list element / + * map entry (as opposed to one row per key), so unlike {@link StateTableMapping} there is no + * per-column projection bookkeeping: column indices in the (fixed) output row are always {@code + * 0=state_key}, {@code 1=list_index/map_key}, {@code 2=list_value/map_value}. + */ +@Internal +public class FlattenedStateTableMapping implements Serializable, SingleColumnStateMapping { + + private static final long serialVersionUID = 1L; + + public static final int STATE_KEY_COLUMN_INDEX = 0; + public static final int SUB_KEY_COLUMN_INDEX = 1; + public static final int VALUE_COLUMN_INDEX = 2; + + private final String stateName; + private final SavepointConnectorOptions.StateType stateType; + private final TypeInformation<?> keyTypeInfo; + @Nullable private final TypeSerializer<?> mapKeyTypeSerializer; + private final TypeSerializer<?> valueTypeSerializer; + @Nullable private StateDescriptor stateDescriptor; + + public FlattenedStateTableMapping( + String stateName, + SavepointConnectorOptions.StateType stateType, + TypeInformation<?> keyTypeInfo, + @Nullable TypeSerializer<?> mapKeyTypeSerializer, + TypeSerializer<?> valueTypeSerializer) { + Preconditions.checkArgument( + stateType == SavepointConnectorOptions.StateType.LIST + || stateType == SavepointConnectorOptions.StateType.MAP, + "Flattened state tables only support LIST and MAP states, got: " + stateType); + this.stateName = stateName; + this.stateType = stateType; + this.keyTypeInfo = keyTypeInfo; + this.mapKeyTypeSerializer = mapKeyTypeSerializer; + this.valueTypeSerializer = valueTypeSerializer; + } + + @Override + public String getStateName() { + return stateName; + } + + @Override + public SavepointConnectorOptions.StateType getStateType() { + return stateType; + } + + @Override + public TypeInformation<?> getKeyTypeInfo() { + return keyTypeInfo; + } + + @Override + @Nullable + public TypeSerializer<?> getMapKeyTypeSerializer() { + return mapKeyTypeSerializer; + } + + @Override + public TypeSerializer<?> getValueTypeSerializer() { + return valueTypeSerializer; + } + + @Override + @SuppressWarnings("rawtypes") + public void setStateDescriptor(StateDescriptor stateDescriptor) { + this.stateDescriptor = stateDescriptor; + } + + @Nullable + @SuppressWarnings("rawtypes") + public StateDescriptor getStateDescriptor() { + return stateDescriptor; + } + + // ------------------------------------------------------------------------- + // Factory + // ------------------------------------------------------------------------- + + /** + * Validates that the table schema matches the fixed 3-column flattened layout with a composite + * primary key on {@code (state_key, list_index/map_key)}, and returns the state type (LIST or + * MAP), inferred from whether the second column is named {@code list_index} or {@code map_key}. + * This is a purely structural check; it performs no I/O or class loading. + */ + public static SavepointConnectorOptions.StateType validateFlattenedSchema( + ResolvedCatalogTable catalogTable) { + ResolvedSchema schema = catalogTable.getResolvedSchema(); + List<Column> columns = schema.getColumns(); + if (columns.size() != 3) { + throw new ValidationException( + "Flattened keyed state tables must have exactly 3 columns " + + "(state_key, list_index/map_key, list_value/map_value), but found " + + columns.size() + + "."); + } + DataType physicalDataType = schema.toPhysicalRowDataType(); + Preconditions.checkArgument( + physicalDataType.getLogicalType().is(LogicalTypeRoot.ROW), + "Row data type expected."); + + String stateKeyColumnName = columns.get(STATE_KEY_COLUMN_INDEX).getName(); + String subKeyColumnName = columns.get(SUB_KEY_COLUMN_INDEX).getName(); + String valueColumnName = columns.get(VALUE_COLUMN_INDEX).getName(); + SavepointConnectorOptions.StateType stateType = + TableMappingSupport.inferFlattenedStateTypeAndValidateValueColumn( + "Flattened keyed state tables", + "second", + "third", + subKeyColumnName, + valueColumnName); + + List<String> expectedKeyColumns = List.of(stateKeyColumnName, subKeyColumnName); + List<String> primaryKeyColumns = + schema.getPrimaryKey().map(UniqueConstraint::getColumns).orElse(List.of()); + if (!primaryKeyColumns.equals(expectedKeyColumns)) { + throw new ValidationException( + "Flattened keyed state tables must declare a composite primary key on (" + + stateKeyColumnName + + ", " + + subKeyColumnName + + "), but found: " + + (primaryKeyColumns.isEmpty() ? "none" : primaryKeyColumns) + + "."); + } + + return stateType; + } + + /** + * Builds a complete {@link FlattenedStateTableMapping}, loading operator state metadata from + * the savepoint and resolving serializers and key type from it. + * + * <p>Assumes {@link #validateFlattenedSchema} has already been called for this table. + * + * <p>This performs I/O (savepoint metadata loading); callers should invoke it lazily, deferred + * to scan time, to keep planning free of savepoint access. + * + * @param catalogTable the resolved table whose schema drives the mapping + * @param stateName the name of the flattened LIST/MAP state, resolved from {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME} + * @param statePath path to the savepoint containing the operator state metadata + * @param operatorIdentifier identifies the operator whose state metadata is loaded + * @param serializerConfig serializer config used when creating serializers from resolved types + * @param stateType {@code LIST} or {@code MAP} + */ + public static FlattenedStateTableMapping from( + ResolvedCatalogTable catalogTable, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier, + SerializerConfig serializerConfig, + SavepointConnectorOptions.StateType stateType) { + + SavepointTypeInfoResolver typeResolver = + TableMappingSupport.createTypeResolver( + statePath, operatorIdentifier, serializerConfig); + + DataType physicalDataType = catalogTable.getResolvedSchema().toPhysicalRowDataType(); + RowType rowType = (RowType) physicalDataType.getLogicalType(); + + TableMappingSupport.FlattenedSerializers serializers = + TableMappingSupport.resolveFlattenedSerializers( + rowType, + typeResolver, + stateName, + stateType, + STATE_KEY_COLUMN_INDEX, + SUB_KEY_COLUMN_INDEX, + VALUE_COLUMN_INDEX); + + return new FlattenedStateTableMapping( + stateName, + stateType, + serializers.keyTypeInfo, + serializers.mapKeyTypeSerializer, + serializers.valueTypeSerializer); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java index b0461bbc127..b3a9c17d1a7 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java @@ -65,6 +65,14 @@ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac stateBackendType, statePath, operatorIdentifier); + case KEYED_FLAT: + return createFlattenedDynamicTableSource( + context, + options, + serializerConfig, + stateBackendType, + statePath, + operatorIdentifier); default: throw new IllegalArgumentException("Unsupported state reader mode: " + readerMode); } @@ -117,6 +125,66 @@ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac SavepointDataStreamScanProvider::new); } + /** + * Creates a {@link FlattenedSavepointDynamicTableSource} for a table exposing a single + * flattened LIST/MAP state (selected via {@link SavepointConnectorOptions#STATE_READER_MODE} + * being set to {@link SavepointConnectorOptions.StateReaderMode#KEYED_FLAT}). The state name is + * resolved from {@link SavepointConnectorOptions#FLATTENED_STATE_NAME}. + */ + private DynamicTableSource createFlattenedDynamicTableSource( + Context context, + Configuration options, + SerializerConfig serializerConfig, + String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier) { + + SavepointConnectorOptions.StateType stateType = + FlattenedStateTableMapping.validateFlattenedSchema(context.getCatalogTable()); + + RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType(); + + String stateName = validateAndGetFlattenedStateName(options); + + // Defer I/O to scan time by creating the mapping lazily. + Supplier<FlattenedStateTableMapping> mappingSupplier = + () -> + FlattenedStateTableMapping.from( + context.getCatalogTable(), + stateName, + statePath, + operatorIdentifier, + serializerConfig, + stateType); + + return new FlattenedSavepointDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + FlattenedStateTableMapping.STATE_KEY_COLUMN_INDEX, + mappingSupplier, + rowType, + "Flattened Savepoint Table Source", + FlattenedSavepointDataStreamScanProvider::new); + } + + /** + * Validates {@code options} against the required/optional option sets extended with {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME}, and returns the resolved state name — shared + * by every table kind whose columns represent a single named state's flattened value fields (or + * a single scalar value column) rather than encoding the state's name via the column layout + * itself. + */ + private String validateAndGetFlattenedStateName(Configuration options) { + Set<ConfigOption<?>> requiredOptions = new HashSet<>(requiredOptions()); + requiredOptions.add(SavepointConnectorOptions.FLATTENED_STATE_NAME); + Set<ConfigOption<?>> optionalOptions = new HashSet<>(optionalOptions()); + + validateOptions(options, requiredOptions, optionalOptions); + + return options.get(SavepointConnectorOptions.FLATTENED_STATE_NAME); + } + /** * Validates {@code options} against the given required/optional option sets and ensures no * unrecognized keys remain (shared by both the general and flattened table source paths). @@ -164,6 +232,11 @@ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac // by StateCatalog. options.add(STATE_READER_MODE); + // Required only for STATE_READER_MODE == KEYED_FLAT/WINDOWED_FLAT (enforced in + // validateAndGetFlattenedStateName); listed here as optional so that generic option + // introspection (docs, Table API tooling) can discover it regardless of mode. + options.add(SavepointConnectorOptions.FLATTENED_STATE_NAME); + return options; } } diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SingleColumnStateMapping.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SingleColumnStateMapping.java new file mode 100644 index 00000000000..e2cbc94366e --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SingleColumnStateMapping.java @@ -0,0 +1,47 @@ +/* + * 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.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.api.common.typeutils.TypeSerializer; + +import javax.annotation.Nullable; + +/** + * Mixed into mapping classes that back a table with exactly one flattened LIST/MAP state, described + * by a single fixed descriptor rather than a list of value columns. Implemented by {@link + * FlattenedStateTableMapping} and {@link WindowFlattenedStateTableMapping}, allowing {@link + * AbstractSingleColumnScanProvider} to build their state descriptor generically. + */ +@Internal +@SuppressWarnings("rawtypes") +interface SingleColumnStateMapping extends SavepointStateMapping { + + String getStateName(); + + SavepointConnectorOptions.StateType getStateType(); + + @Nullable + TypeSerializer getMapKeyTypeSerializer(); + + TypeSerializer getValueTypeSerializer(); + + void setStateDescriptor(StateDescriptor stateDescriptor); +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java index 8df34fb225d..52272e698ea 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java @@ -19,6 +19,8 @@ package org.apache.flink.state.table; import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ConfigOptions; @@ -26,13 +28,18 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.state.api.OperatorIdentifier; import org.apache.flink.state.api.runtime.SavepointLoader; import org.apache.flink.state.api.runtime.SavepointLoader.OperatorStateMetadata; +import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.ArrayType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.MapType; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; import org.apache.flink.util.Preconditions; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.List; import java.util.stream.IntStream; @@ -167,19 +174,176 @@ final class TableMappingSupport { fieldOption(valueRowField.getName(), STATE_NAME).stringType().noDefaultValue(); String stateName = options.getOptional(stateNameOption).orElse(valueRowField.getName()); - SavepointConnectorOptions.StateType stateType = inferStateType(valueRowField.getType()); + StateDescriptor.Type actualStateKind = typeResolver.resolveStateKind(stateName); + SavepointConnectorOptions.StateType stateType = + stateTypeFromActualKind(actualStateKind, valueRowField.getType()); TypeSerializer mapKeyTypeSerializer = typeResolver.resolveMapKeySerializer( valueRowField, stateType == SavepointConnectorOptions.StateType.MAP); - TypeSerializer valueTypeSerializer = typeResolver.resolveValueSerializer(valueRowField); + // A VALUE-shaped state (including REDUCING/AGGREGATING) whose value happens to be a + // List/Map (e.g. a collect-list accumulator) must resolve the serializer for the whole + // value, not the element/entry serializer resolveValueSerializer would unwrap it to for a + // genuine keyed LIST/MAP state. + TypeSerializer valueTypeSerializer = + stateType == SavepointConnectorOptions.StateType.VALUE + ? typeResolver.resolveFlatValueSerializer(valueRowField) + : typeResolver.resolveValueSerializer(valueRowField); return new StateValueColumnConfiguration( columnIndex, stateName, stateType, - typeResolver.resolveStateKind(stateName), + actualStateKind, mapKeyTypeSerializer, valueTypeSerializer); } + + /** + * Determines the coarse VALUE/LIST/MAP shape used for the SQL schema, preferring the actual + * {@link StateDescriptor.Type} the state was registered under (resolved from the savepoint + * metadata) over inferring it from the column's SQL type. Without this, a {@code + * ValueState}/{@code ReducingState}/{@code AggregatingState} whose value happens to be a + * List/Map (e.g. a collect-list accumulator) would be misclassified as a keyed LIST/MAP state + * purely because its value's SQL type is ARRAY/MAP, causing {@code KeyedStateReader} to open it + * with the wrong state-getter ({@code getListState}/{@code getMapState} instead of {@code + * getState}/{@code getReducingState}/{@code getAggregatingState}) against state that is + * physically stored in an incompatible format. + * + * <p>Falls back to {@link #inferStateType} when the state is absent from the preloaded metadata + * (i.e. {@code actualStateKind == UNKNOWN}). + */ + private static SavepointConnectorOptions.StateType stateTypeFromActualKind( + StateDescriptor.Type actualStateKind, LogicalType logicalType) { + switch (actualStateKind) { + case LIST: + return SavepointConnectorOptions.StateType.LIST; + case MAP: + return SavepointConnectorOptions.StateType.MAP; + case VALUE: + case REDUCING: + case AGGREGATING: + case FOLDING: + return SavepointConnectorOptions.StateType.VALUE; + default: + return inferStateType(logicalType); + } + } + + /** + * Infers the flattened state type (LIST or MAP) from the sub-key column name and validates that + * the value column is named consistently with it. + * + * @param tableKindLabel e.g. "Flattened keyed state tables", used in validation error messages + * @param subKeyColumnOrdinal ordinal word for the sub-key column's position, e.g. "second" + * @param valueColumnOrdinal ordinal word for the value column's position, e.g. "third" + */ + static SavepointConnectorOptions.StateType inferFlattenedStateTypeAndValidateValueColumn( + String tableKindLabel, + String subKeyColumnOrdinal, + String valueColumnOrdinal, + String subKeyColumnName, + String valueColumnName) { + SavepointConnectorOptions.StateType stateType; + String expectedValueColumnName; + switch (subKeyColumnName) { + case "list_index": + stateType = SavepointConnectorOptions.StateType.LIST; + expectedValueColumnName = "list_value"; + break; + case "map_key": + stateType = SavepointConnectorOptions.StateType.MAP; + expectedValueColumnName = "map_value"; + break; + default: + throw new ValidationException( + tableKindLabel + + " must name their " + + subKeyColumnOrdinal + + " column either 'list_index' (LIST state) or 'map_key' (MAP " + + "state), but found '" + + subKeyColumnName + + "'."); + } + + if (!expectedValueColumnName.equals(valueColumnName)) { + throw new ValidationException( + tableKindLabel + + " must name their " + + valueColumnOrdinal + + " column '" + + expectedValueColumnName + + "', but found '" + + valueColumnName + + "'."); + } + + return stateType; + } + + /** Resolved key type and value-related serializers for a flattened (LIST/MAP) state mapping. */ + static final class FlattenedSerializers { + final TypeInformation<?> keyTypeInfo; + @Nullable final TypeSerializer<?> mapKeyTypeSerializer; + final TypeSerializer<?> valueTypeSerializer; + + FlattenedSerializers( + TypeInformation<?> keyTypeInfo, + @Nullable TypeSerializer<?> mapKeyTypeSerializer, + TypeSerializer<?> valueTypeSerializer) { + this.keyTypeInfo = keyTypeInfo; + this.mapKeyTypeSerializer = mapKeyTypeSerializer; + this.valueTypeSerializer = valueTypeSerializer; + } + } + + /** + * Resolves the key type and value-related serializers shared by {@link + * FlattenedStateTableMapping} and {@link WindowFlattenedStateTableMapping}'s {@code from(...)} + * factories. + */ + static FlattenedSerializers resolveFlattenedSerializers( + RowType rowType, + SavepointTypeInfoResolver typeResolver, + String stateName, + SavepointConnectorOptions.StateType stateType, + int stateKeyColumnIndex, + int subKeyColumnIndex, + int valueColumnIndex) { + TypeInformation<?> keyTypeInfo = + typeResolver.resolveKeyType(rowType.getFields().get(stateKeyColumnIndex)); + + RowType.RowField compositeValueField = + buildCompositeValueField( + rowType, stateName, stateType, subKeyColumnIndex, valueColumnIndex); + + TypeSerializer<?> mapKeyTypeSerializer = + typeResolver.resolveMapKeySerializer( + compositeValueField, stateType == SavepointConnectorOptions.StateType.MAP); + TypeSerializer<?> valueTypeSerializer = + typeResolver.resolveValueSerializer(compositeValueField); + + return new FlattenedSerializers(keyTypeInfo, mapKeyTypeSerializer, valueTypeSerializer); + } + + /** + * Builds a synthetic {@link RowType.RowField} for a flattened LIST/MAP state's composite + * (ArrayType/MapType) value, keyed by {@code stateName} so metadata lookup succeeds, mirroring + * how the general (non-flattened) path resolves value columns. + */ + private static RowType.RowField buildCompositeValueField( + RowType rowType, + String stateName, + SavepointConnectorOptions.StateType stateType, + int subKeyColumnIndex, + int valueColumnIndex) { + LogicalType valueLogicalType = rowType.getFields().get(valueColumnIndex).getType(); + LogicalType compositeLogicalType = + stateType == SavepointConnectorOptions.StateType.LIST + ? new ArrayType(valueLogicalType) + : new MapType( + rowType.getFields().get(subKeyColumnIndex).getType(), + valueLogicalType); + return new RowType.RowField(stateName, compositeLogicalType); + } } diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/StateCatalogDiscoveryITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/StateCatalogDiscoveryITCase.java index 92d0bc256dd..a392bea2806 100644 --- a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/StateCatalogDiscoveryITCase.java +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/catalog/StateCatalogDiscoveryITCase.java @@ -48,8 +48,9 @@ import static org.assertj.core.api.Assertions.assertThat; /** * Integration tests for basic {@link StateCatalog} functionality driven through {@code CREATE * CATALOG} DDL and SQL: multi-label discovery and the {@code metadata} view. Checkpoint metadata is - * written directly via {@link Checkpoints#storeCheckpointMetadata} — no minicluster or real state - * backend is involved, so these tests are backend-agnostic by construction. + * written directly via {@link Checkpoints#storeCheckpointMetadataWithoutExclusiveDir} — no + * minicluster or real state backend is involved, so these tests are backend-agnostic by + * construction. * * <p>For reads of real (generated) keyed-state savepoints, see {@code * StateCatalogGeneratedSavepointITCase} (HashMap-only, checked-in fixtures) and {@code @@ -135,7 +136,8 @@ class StateCatalogDiscoveryITCase { assertThat(catalog.databaseExists(dbName)).isTrue(); assertThat(catalog.databaseExists("app/nonexistent")).isFalse(); - assertThat(catalog.listTables(dbName)).isEmpty(); + // listTables includes views (the "metadata" view), per the Catalog contract. + assertThat(catalog.listTables(dbName)).containsExactly(StateCatalog.METADATA_TABLE); assertThat(catalog.listViews(dbName)).containsExactly(StateCatalog.METADATA_TABLE); assertThat(catalog.tableExists(new ObjectPath(dbName, StateCatalog.METADATA_TABLE))) @@ -193,7 +195,7 @@ class StateCatalogDiscoveryITCase { CheckpointMetadata metadata = new CheckpointMetadata(checkpointId, operators, Collections.emptyList()); try (OutputStream out = Files.newOutputStream(snapshotDir.resolve("_metadata"))) { - Checkpoints.storeCheckpointMetadata(metadata, out); + Checkpoints.storeCheckpointMetadataWithoutExclusiveDir(metadata, out); } }
