This is an automated email from the ASF dual-hosted git repository.
gaborgsomogyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new a4eb67642a8 [FLINK-36929][table] Add SQL connector for keyed savepoint
data
a4eb67642a8 is described below
commit a4eb67642a826b941b02c7221846840326c3ff55
Author: Gabor Somogyi <[email protected]>
AuthorDate: Wed Feb 5 10:14:33 2025 +0100
[FLINK-36929][table] Add SQL connector for keyed savepoint data
---
flink-libraries/flink-state-processing-api/pom.xml | 16 +-
.../apache/flink/state/table/KeyedStateReader.java | 376 +++++++++++++++++++++
.../state/table/SavepointConnectorOptions.java | 139 ++++++++
.../state/table/SavepointConnectorOptionsUtil.java | 49 +++
.../table/SavepointDataStreamScanProvider.java | 139 ++++++++
.../state/table/SavepointDynamicTableSource.java | 85 +++++
.../table/SavepointDynamicTableSourceFactory.java | 355 +++++++++++++++++++
.../state/table/StateValueColumnConfiguration.java | 79 +++++
.../org.apache.flink.table.factories.Factory | 16 +
.../example/state/writer/job/schema/PojoData.java | 33 ++
.../table/SavepointDynamicTableSourceTest.java | 172 ++++++++++
.../src/test/resources/table-state-nulls/_metadata | Bin 0 -> 24017 bytes
.../src/test/resources/table-state/_metadata | Bin 0 -> 22390 bytes
13 files changed, 1458 insertions(+), 1 deletion(-)
diff --git a/flink-libraries/flink-state-processing-api/pom.xml
b/flink-libraries/flink-state-processing-api/pom.xml
index 8e087770421..691c30a9ef4 100644
--- a/flink-libraries/flink-state-processing-api/pom.xml
+++ b/flink-libraries/flink-state-processing-api/pom.xml
@@ -53,7 +53,14 @@ under the License.
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
-
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-runtime</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
<!-- test dependencies -->
<dependency>
@@ -78,6 +85,13 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+
<artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime</artifactId>
diff --git
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/KeyedStateReader.java
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/KeyedStateReader.java
new file mode 100644
index 00000000000..18279448512
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/KeyedStateReader.java
@@ -0,0 +1,376 @@
+/*
+ * 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.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.state.api.functions.KeyedStateReaderFunction;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.Collector;
+
+import org.apache.flink.shaded.guava32.com.google.common.cache.Cache;
+import org.apache.flink.shaded.guava32.com.google.common.cache.CacheBuilder;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.StreamSupport;
+
+/** Keyed state reader function for value, list and map state types. */
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class KeyedStateReader extends KeyedStateReaderFunction<Object,
RowData> {
+ private static final long CACHE_MAX_SIZE = 64000L;
+
+ private final Tuple2<Integer, List<StateValueColumnConfiguration>>
keyValueProjections;
+ private final RowType rowType;
+ private final Map<Integer, State> states = new HashMap<>();
+ private final Cache<Tuple2<Class, String>, Field> classFieldCache;
+ private final Cache<Tuple2<Class, String>, Method> classMethodCache;
+
+ public KeyedStateReader(
+ final RowType rowType,
+ final Tuple2<Integer, List<StateValueColumnConfiguration>>
keyValueProjections) {
+ this.keyValueProjections = keyValueProjections;
+ this.rowType = rowType;
+ this.classMethodCache =
CacheBuilder.newBuilder().maximumSize(CACHE_MAX_SIZE).build();
+ this.classFieldCache =
CacheBuilder.newBuilder().maximumSize(CACHE_MAX_SIZE).build();
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ for (StateValueColumnConfiguration columnConfig :
keyValueProjections.f1) {
+ switch (columnConfig.getStateType()) {
+ case VALUE:
+ states.put(
+ columnConfig.getColumnIndex(),
+ getRuntimeContext()
+ .getState(
+ (ValueStateDescriptor)
+
columnConfig.getStateDescriptor()));
+ break;
+
+ case LIST:
+ states.put(
+ columnConfig.getColumnIndex(),
+ getRuntimeContext()
+ .getListState(
+ (ListStateDescriptor)
+
columnConfig.getStateDescriptor()));
+ break;
+
+ case MAP:
+ states.put(
+ columnConfig.getColumnIndex(),
+ getRuntimeContext()
+ .getMapState(
+ (MapStateDescriptor)
+
columnConfig.getStateDescriptor()));
+ break;
+
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported state type: " +
columnConfig.getStateType());
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ states.clear();
+ }
+
+ @Override
+ public void readKey(Object key, Context context, Collector<RowData> out)
throws Exception {
+ GenericRowData row = new GenericRowData(RowKind.INSERT, 1 +
keyValueProjections.f1.size());
+
+ List<RowType.RowField> fields = rowType.getFields();
+
+ // Fill column from key
+ int columnIndex = keyValueProjections.f0;
+ LogicalType keyLogicalType = fields.get(columnIndex).getType();
+ row.setField(columnIndex, getValue(keyLogicalType, key));
+
+ // Fill columns from values
+ for (StateValueColumnConfiguration columnConfig :
keyValueProjections.f1) {
+ LogicalType valueLogicalType =
fields.get(columnConfig.getColumnIndex()).getType();
+ switch (columnConfig.getStateType()) {
+ case VALUE:
+ row.setField(
+ columnConfig.getColumnIndex(),
+ getValue(
+ valueLogicalType,
+ ((ValueState)
states.get(columnConfig.getColumnIndex()))
+ .value()));
+ break;
+
+ case LIST:
+ row.setField(
+ columnConfig.getColumnIndex(),
+ getValue(
+ valueLogicalType,
+ ((ListState)
states.get(columnConfig.getColumnIndex())).get()));
+ break;
+
+ case MAP:
+ row.setField(
+ columnConfig.getColumnIndex(),
+ getValue(
+ valueLogicalType,
+ ((MapState)
states.get(columnConfig.getColumnIndex()))
+ .entries()));
+ break;
+
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported state type: " +
columnConfig.getStateType());
+ }
+ }
+
+ out.collect(row);
+ }
+
+ private Object getValue(LogicalType logicalType, Object object) {
+ if (object == null) {
+ return null;
+ }
+ switch (logicalType.getTypeRoot()) {
+ case CHAR: // String
+ case VARCHAR: // String
+ return StringData.fromString(object.toString());
+
+ case BOOLEAN: // Boolean
+ return object;
+
+ case BINARY: // byte[]
+ case VARBINARY: // ByteBuffer, byte[]
+ return convertToBytes(object);
+
+ case DECIMAL: // BigDecimal, ByteBuffer, byte[]
+ return convertToDecimal(object, logicalType);
+
+ case TINYINT: // Byte
+ case SMALLINT: // Short
+ case INTEGER: // Integer
+ case BIGINT: // Long
+ case FLOAT: // Float
+ case DOUBLE: // Double
+ case DATE: // Integer
+ return object;
+
+ case INTERVAL_YEAR_MONTH: // Long
+ case INTERVAL_DAY_TIME: // Long
+ return object;
+
+ case ARRAY:
+ return convertToArray(object, logicalType);
+
+ case MAP:
+ return convertToMap(object, logicalType);
+
+ case ROW:
+ return convertToRow(object, logicalType);
+
+ case NULL:
+ return null;
+
+ case MULTISET:
+ case TIME_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ case DISTINCT_TYPE:
+ case STRUCTURED_TYPE:
+ case RAW:
+ case SYMBOL:
+ case UNRESOLVED:
+ default:
+ throw new UnsupportedOperationException("Unsupported type: " +
logicalType);
+ }
+ }
+
+ private Object getObjectField(Object object, RowType.RowField rowField) {
+ String rowFieldName = rowField.getName();
+
+ Class objectClass = object.getClass();
+ Object objectField;
+ try {
+ Field field =
+ classFieldCache.get(
+ Tuple2.of(objectClass, rowFieldName),
+ () -> objectClass.getField(rowFieldName));
+ objectField = field.get(object);
+ } catch (ExecutionException e1) {
+ Method method = getMethod(objectClass, rowFieldName);
+ try {
+ objectField = method.invoke(object);
+ } catch (IllegalAccessException | InvocationTargetException e2) {
+ throw new RuntimeException(e2);
+ }
+ } catch (IllegalAccessException e) {
+ throw new UnsupportedOperationException(
+ "Cannot access field by either public member or getter
function: "
+ + rowField.getName());
+ }
+
+ return objectField;
+ }
+
+ private Method getMethod(Class objectClass, String rowFieldName) {
+ String upperRowFieldName =
+ rowFieldName.substring(0, 1).toUpperCase() +
rowFieldName.substring(1);
+ try {
+ String methodName = "get" + upperRowFieldName;
+ return classMethodCache.get(
+ Tuple2.of(objectClass, methodName), () ->
objectClass.getMethod(methodName));
+ } catch (ExecutionException e1) {
+ try {
+ String methodName = "is" + upperRowFieldName;
+ return classMethodCache.get(
+ Tuple2.of(objectClass, methodName),
+ () -> objectClass.getMethod(methodName));
+ } catch (ExecutionException e2) {
+ throw new RuntimeException(e2);
+ }
+ }
+ }
+
+ private static DecimalData convertToDecimal(Object object, LogicalType
logicalType) {
+ DecimalType decimalType = (DecimalType) logicalType;
+
+ final int precision = decimalType.getPrecision();
+ final int scale = decimalType.getScale();
+ if (object instanceof BigDecimal) {
+ return DecimalData.fromBigDecimal((BigDecimal) object, precision,
scale);
+ } else if (object instanceof ByteBuffer) {
+ ByteBuffer byteBuffer = (ByteBuffer) object;
+ final byte[] bytes = new byte[byteBuffer.remaining()];
+ byteBuffer.get(bytes);
+ return DecimalData.fromUnscaledBytes(bytes, precision, scale);
+ } else if (object instanceof byte[]) {
+ final byte[] bytes = (byte[]) object;
+ return DecimalData.fromUnscaledBytes(bytes, precision, scale);
+ } else {
+ throw new UnsupportedOperationException(
+ "Decimal conversion supports only BigDecimal, ByteBuffer
and byte[] but received: "
+ + object.getClass().getName());
+ }
+ }
+
+ private byte[] convertToBytes(Object object) {
+ if (object instanceof ByteBuffer) {
+ ByteBuffer byteBuffer = (ByteBuffer) object;
+ byte[] bytes = new byte[byteBuffer.remaining()];
+ byteBuffer.get(bytes);
+ return bytes;
+ } else if (object instanceof byte[]) {
+ return (byte[]) object;
+ } else {
+ throw new UnsupportedOperationException(
+ "Byte array conversion supports only ByteBuffer and byte[]
but received: "
+ + object.getClass().getName());
+ }
+ }
+
+ private GenericArrayData convertToArray(Object object, LogicalType
logicalType) {
+ LogicalType elementLogicalType = ((ArrayType)
logicalType).getElementType();
+
+ if (object instanceof Iterable) {
+ Iterable iterable = (Iterable) object;
+ return new GenericArrayData(
+ StreamSupport.stream(iterable.spliterator(), false)
+ .map(v -> getValue(elementLogicalType, v))
+ .toArray());
+ } else {
+ throw new UnsupportedOperationException(
+ "Array conversion supports only Iterable but received: "
+ + object.getClass().getName());
+ }
+ }
+
+ private GenericMapData convertToMap(Object object, LogicalType
logicalType) {
+ MapType mapType = (MapType) logicalType;
+ LogicalType keyLogicalType = mapType.getKeyType();
+ LogicalType valueLogicalType = mapType.getValueType();
+
+ if (object instanceof Iterable) {
+ Iterable iterable = (Iterable) object;
+ Iterator iterator = iterable.iterator();
+ Map<Object, Object> result = new HashMap<>();
+ boolean typeChecked = false;
+ while (iterator.hasNext()) {
+ Object e = iterator.next();
+ // The boolean check here is for performance tuning because
instanceof is slow, and
+ // it's enough to check the type only once.
+ if (!typeChecked && !(e instanceof Map.Entry)) {
+ throw new UnsupportedOperationException(
+ "Map conversion supports only Iterable<Map.Entry>
but received: "
+ + object.getClass().getName());
+ } else {
+ typeChecked = true;
+ }
+ Map.Entry entry = (Map.Entry) e;
+ result.put(
+ getValue(keyLogicalType, entry.getKey()),
+ getValue(valueLogicalType, entry.getValue()));
+ }
+ return new GenericMapData(result);
+ } else {
+ throw new UnsupportedOperationException(
+ "Map conversion supports only Iterable<Map.Entry> but
received: "
+ + object.getClass().getName());
+ }
+ }
+
+ private GenericRowData convertToRow(Object object, LogicalType
logicalType) {
+ RowType rowType = (RowType) logicalType;
+ GenericRowData result = new GenericRowData(RowKind.INSERT,
rowType.getFieldCount());
+ List<RowType.RowField> fields = rowType.getFields();
+ for (int i = 0; i < rowType.getFieldCount(); i++) {
+ RowType.RowField subRowField = fields.get(i);
+ Object subObject = getObjectField(object, subRowField);
+ result.setField(i, getValue(subRowField.getType(), subObject));
+ }
+ return result;
+ }
+}
diff --git
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointConnectorOptions.java
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointConnectorOptions.java
new file mode 100644
index 00000000000..1c596ad4b49
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointConnectorOptions.java
@@ -0,0 +1,139 @@
+/*
+ * 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.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.configuration.description.TextElement;
+
+import static org.apache.flink.configuration.description.TextElement.code;
+
+/** Options for the state connector. */
+@PublicEvolving
+public class SavepointConnectorOptions {
+
+ public static final String FIELDS = "fields";
+ public static final String STATE_NAME = "state-name";
+ public static final String STATE_TYPE = "state-type";
+ public static final String MAP_KEY_FORMAT = "map-key-format";
+ public static final String VALUE_FORMAT = "value-format";
+
+ /** Value state types. */
+ public enum StateType {
+ VALUE,
+ LIST,
+ MAP
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // Common options
+ //
--------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> STATE_BACKEND_TYPE =
+ ConfigOptions.key("state.backend.type")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("The state backend to be used to
read state.")
+ .linebreak()
+ .text(
+ "The implementation can be
specified either via their shortcut "
+ + " name, or via the class
name of a %s. "
+ + "If a factory is
specified it is instantiated via its "
+ + "zero argument
constructor and its %s "
+ + "method is called.",
+
TextElement.code("StateBackendFactory"),
+ TextElement.code(
+
"StateBackendFactory#createFromConfig(ReadableConfig, ClassLoader)"))
+ .linebreak()
+ .text(
+ "Recognized shortcut names are
'hashmap', 'rocksdb' and 'forst'.")
+ .build());
+
+ public static final ConfigOption<String> STATE_PATH =
+ ConfigOptions.key("state.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Defines the state path which must be used for
state reading.");
+
+ public static final ConfigOption<String> OPERATOR_UID =
+ ConfigOptions.key("operator.uid")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Defines the operator UID which must be used for
state reading (Can't be used together with UID hash).");
+
+ public static final ConfigOption<String> OPERATOR_UID_HASH =
+ ConfigOptions.key("operator.uid.hash")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Defines the operator UID hash which must be used
for state reading (Can't be used together with UID).");
+
+ //
--------------------------------------------------------------------------------------------
+ // Value options
+ //
--------------------------------------------------------------------------------------------
+
+ /** Placeholder {@link ConfigOption}. Not used for retrieving values. */
+ public static final ConfigOption<String> STATE_NAME_PLACEHOLDER =
+ ConfigOptions.key(String.format("%s.#.%s", FIELDS, STATE_NAME))
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Defines the state name which must be used for
state reading.");
+
+ /** Placeholder {@link ConfigOption}. Not used for retrieving values. */
+ public static final ConfigOption<StateType> STATE_TYPE_PLACEHOLDER =
+ ConfigOptions.key(String.format("%s.#.%s", FIELDS, STATE_TYPE))
+ .enumType(StateType.class)
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Defines the state type which must
be used for state reading, including %s, %s and %s. "
+ + "When it's not provided
then it tries to be inferred from the SQL type (ARRAY=list, MAP=map, all
others=value).",
+ code(StateType.VALUE.toString()),
+ code(StateType.LIST.toString()),
+ code(StateType.MAP.toString()))
+ .build());
+
+ /** Placeholder {@link ConfigOption}. Not used for retrieving values. */
+ public static final ConfigOption<String> MAP_KEY_FORMAT_PLACEHOLDER =
+ ConfigOptions.key(String.format("%s.#.%s", FIELDS, MAP_KEY_FORMAT))
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Defines the format class scheme for decoding map
value key data. "
+ + "When it's not provided then it tries to
be inferred from the SQL type (only primitive types supported).");
+
+ /** Placeholder {@link ConfigOption}. Not used for retrieving values. */
+ public static final ConfigOption<String> VALUE_FORMAT_PLACEHOLDER =
+ ConfigOptions.key(String.format("%s.#.%s", FIELDS, VALUE_FORMAT))
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Defines the format class scheme for decoding
value data. "
+ + "When it's not provided then it tries to
be inferred from the SQL type (only primitive types supported).");
+
+ private SavepointConnectorOptions() {}
+}
diff --git
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointConnectorOptionsUtil.java
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointConnectorOptionsUtil.java
new file mode 100644
index 00000000000..bdaba67411b
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointConnectorOptionsUtil.java
@@ -0,0 +1,49 @@
+/*
+ * 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.PublicEvolving;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.state.api.OperatorIdentifier;
+
+import java.util.Optional;
+
+import static
org.apache.flink.state.table.SavepointConnectorOptions.OPERATOR_UID;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.OPERATOR_UID_HASH;
+
+/** Utilities for {@link SavepointConnectorOptions}. */
+@PublicEvolving
+public class SavepointConnectorOptionsUtil {
+
+ public static OperatorIdentifier getOperatorIdentifier(ReadableConfig
options) {
+ final Optional<String> operatorUid = options.getOptional(OPERATOR_UID);
+ final Optional<String> operatorUidHash =
options.getOptional(OPERATOR_UID_HASH);
+
+ if (operatorUid.isPresent() == operatorUidHash.isPresent()) {
+ throw new IllegalArgumentException(
+ "Either operator uid or operator uid hash must be
specified.");
+ }
+
+ return operatorUid
+ .map(OperatorIdentifier::forUid)
+ .orElseGet(() ->
OperatorIdentifier.forUidHash(operatorUidHash.get()));
+ }
+
+ private SavepointConnectorOptionsUtil() {}
+}
diff --git
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDataStreamScanProvider.java
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDataStreamScanProvider.java
new file mode 100644
index 00000000000..e71bff21329
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDataStreamScanProvider.java
@@ -0,0 +1,139 @@
+/*
+ * 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.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.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.StateBackendOptions;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.runtime.state.StateBackendLoader;
+import org.apache.flink.state.api.OperatorIdentifier;
+import org.apache.flink.state.api.SavepointReader;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.connector.ProviderContext;
+import org.apache.flink.table.connector.source.DataStreamScanProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+
+import javax.naming.ConfigurationException;
+
+import java.util.List;
+
+/** State data stream scan provider. */
+public class SavepointDataStreamScanProvider implements DataStreamScanProvider
{
+ private final String stateBackendType;
+ private final String statePath;
+ private final OperatorIdentifier operatorIdentifier;
+ private final String keyFormat;
+ private final Tuple2<Integer, List<StateValueColumnConfiguration>>
keyValueProjections;
+ private final RowType rowType;
+
+ public SavepointDataStreamScanProvider(
+ final String stateBackendType,
+ final String statePath,
+ final OperatorIdentifier operatorIdentifier,
+ final String keyFormat,
+ final Tuple2<Integer, List<StateValueColumnConfiguration>>
keyValueProjections,
+ RowType rowType) {
+ this.stateBackendType = stateBackendType;
+ this.statePath = statePath;
+ this.operatorIdentifier = operatorIdentifier;
+ this.keyFormat = keyFormat;
+ this.keyValueProjections = keyValueProjections;
+ this.rowType = rowType;
+ }
+
+ @Override
+ public boolean isBounded() {
+ return true;
+ }
+
+ @Override
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ public DataStream<RowData> produceDataStream(
+ ProviderContext providerContext, StreamExecutionEnvironment
execEnv) {
+ try {
+ Configuration configuration = new Configuration();
+ configuration.set(StateBackendOptions.STATE_BACKEND,
stateBackendType);
+ StateBackend stateBackend =
+ StateBackendLoader.loadStateBackendFromConfig(
+ configuration, getClass().getClassLoader(), null);
+
+ SavepointReader savepointReader =
+ SavepointReader.read(execEnv, statePath, stateBackend);
+
+ // Get key type information
+ TypeInformation keyTypeInfo =
TypeInformation.of(Class.forName(keyFormat));
+
+ // Get value state descriptors
+ for (StateValueColumnConfiguration columnConfig :
keyValueProjections.f1) {
+ TypeInformation valueTypeInfo =
+
TypeInformation.of(Class.forName(columnConfig.getValueFormat()));
+
+ switch (columnConfig.getStateType()) {
+ case VALUE:
+ columnConfig.setStateDescriptor(
+ new ValueStateDescriptor<>(
+ columnConfig.getStateName(),
valueTypeInfo));
+ break;
+
+ case LIST:
+ columnConfig.setStateDescriptor(
+ new ListStateDescriptor<>(
+ columnConfig.getStateName(),
valueTypeInfo));
+ break;
+
+ case MAP:
+ if (columnConfig.getMapKeyFormat() == null) {
+ throw new ConfigurationException(
+ "Map key format is required for map
state");
+ }
+ TypeInformation<?> mapKeyTypeInfo =
+
TypeInformation.of(Class.forName(columnConfig.getMapKeyFormat()));
+ columnConfig.setStateDescriptor(
+ new MapStateDescriptor<>(
+ columnConfig.getStateName(),
+ mapKeyTypeInfo,
+ valueTypeInfo));
+ break;
+
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported state type: " +
columnConfig.getStateType());
+ }
+ }
+
+ TypeInformation outTypeInfo = InternalTypeInfo.of(rowType);
+
+ return savepointReader.readKeyedState(
+ operatorIdentifier,
+ new KeyedStateReader(rowType, keyValueProjections),
+ keyTypeInfo,
+ outTypeInfo);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSource.java
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSource.java
new file mode 100644
index 00000000000..e22b3ea9f5a
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSource.java
@@ -0,0 +1,85 @@
+/*
+ * 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.java.tuple.Tuple2;
+import org.apache.flink.state.api.OperatorIdentifier;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.List;
+
+/** State dynamic source. */
+public class SavepointDynamicTableSource implements ScanTableSource {
+ private final String stateBackendType;
+ private final String statePath;
+ private final OperatorIdentifier operatorIdentifier;
+ private final String keyFormat;
+ private final Tuple2<Integer, List<StateValueColumnConfiguration>>
keyValueProjections;
+ private final RowType rowType;
+
+ public SavepointDynamicTableSource(
+ final String stateBackendType,
+ final String statePath,
+ final OperatorIdentifier operatorIdentifier,
+ final String keyFormat,
+ final Tuple2<Integer, List<StateValueColumnConfiguration>>
keyValueProjections,
+ RowType rowType) {
+ this.stateBackendType = stateBackendType;
+ this.statePath = statePath;
+ this.operatorIdentifier = operatorIdentifier;
+ this.keyValueProjections = keyValueProjections;
+ this.keyFormat = keyFormat;
+ this.rowType = rowType;
+ }
+
+ @Override
+ public DynamicTableSource copy() {
+ return new SavepointDynamicTableSource(
+ stateBackendType,
+ statePath,
+ operatorIdentifier,
+ keyFormat,
+ keyValueProjections,
+ rowType);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "State Table Source";
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext)
{
+ return new SavepointDataStreamScanProvider(
+ stateBackendType,
+ statePath,
+ operatorIdentifier,
+ keyFormat,
+ keyValueProjections,
+ rowType);
+ }
+}
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
new file mode 100644
index 00000000000..8cdbb1a2c99
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java
@@ -0,0 +1,355 @@
+/*
+ * 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.java.tuple.Tuple2;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.state.api.OperatorIdentifier;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+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.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.math.BigDecimal;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+import static org.apache.flink.state.table.SavepointConnectorOptions.FIELDS;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.MAP_KEY_FORMAT;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.MAP_KEY_FORMAT_PLACEHOLDER;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.OPERATOR_UID;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.OPERATOR_UID_HASH;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.STATE_BACKEND_TYPE;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.STATE_NAME;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.STATE_NAME_PLACEHOLDER;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.STATE_PATH;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.STATE_TYPE;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.STATE_TYPE_PLACEHOLDER;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.VALUE_FORMAT;
+import static
org.apache.flink.state.table.SavepointConnectorOptions.VALUE_FORMAT_PLACEHOLDER;
+import static
org.apache.flink.state.table.SavepointConnectorOptionsUtil.getOperatorIdentifier;
+import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR;
+
+/** Dynamic source factory for {@link SavepointDynamicTableSource}. */
+public class SavepointDynamicTableSourceFactory implements
DynamicTableSourceFactory {
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ Configuration options = new Configuration();
+ context.getCatalogTable().getOptions().forEach(options::setString);
+
+ final String stateBackendType = options.get(STATE_BACKEND_TYPE);
+ final String statePath = options.get(STATE_PATH);
+ final OperatorIdentifier operatorIdentifier =
getOperatorIdentifier(options);
+
+ final Tuple2<Integer, int[]> keyValueProjections =
+ createKeyValueProjections(context.getCatalogTable());
+
+ LogicalType logicalType =
context.getPhysicalRowDataType().getLogicalType();
+ Preconditions.checkArgument(logicalType.is(LogicalTypeRoot.ROW), "Row
data type expected.");
+ RowType rowType = (RowType) logicalType;
+
+ Set<ConfigOption<?>> requiredOptions = new
HashSet<>(requiredOptions());
+ Set<ConfigOption<?>> optionalOptions = new
HashSet<>(optionalOptions());
+
+ RowType.RowField keyRowField =
rowType.getFields().get(keyValueProjections.f0);
+ ConfigOption<String> keyFormatOption =
+ key(String.format("%s.%s.%s", FIELDS, keyRowField.getName(),
VALUE_FORMAT))
+ .stringType()
+ .noDefaultValue();
+ optionalOptions.add(keyFormatOption);
+ final String keyFormat =
+ options.getOptional(keyFormatOption)
+ .orElseGet(
+ () ->
+ inferStateValueFormat(
+ keyRowField.getName(),
keyRowField.getType()));
+
+ final Tuple2<Integer, List<StateValueColumnConfiguration>>
keyValueConfigProjections =
+ Tuple2.of(
+ keyValueProjections.f0,
+ Arrays.stream(keyValueProjections.f1)
+ .mapToObj(
+ columnIndex -> {
+ RowType.RowField valueRowField =
+
rowType.getFields().get(columnIndex);
+
+ ConfigOption<String>
stateNameOption =
+ key(String.format(
+ "%s.%s.%s",
+ FIELDS,
+
valueRowField.getName(),
+
STATE_NAME))
+ .stringType()
+ .noDefaultValue();
+
optionalOptions.add(stateNameOption);
+
+
ConfigOption<SavepointConnectorOptions.StateType>
+ stateTypeOption =
+ key(String.format(
+
"%s.%s.%s",
+
FIELDS,
+
valueRowField.getName(),
+
STATE_TYPE))
+ .enumType(
+
SavepointConnectorOptions
+
.StateType
+
.class)
+
.noDefaultValue();
+
optionalOptions.add(stateTypeOption);
+
+ ConfigOption<String>
mapKeyFormatOption =
+ key(String.format(
+ "%s.%s.%s",
+ FIELDS,
+
valueRowField.getName(),
+
MAP_KEY_FORMAT))
+ .stringType()
+ .noDefaultValue();
+
optionalOptions.add(mapKeyFormatOption);
+
+ ConfigOption<String>
valueFormatOption =
+ key(String.format(
+ "%s.%s.%s",
+ FIELDS,
+
valueRowField.getName(),
+
VALUE_FORMAT))
+ .stringType()
+ .noDefaultValue();
+
optionalOptions.add(valueFormatOption);
+
+ LogicalType valueLogicalType =
valueRowField.getType();
+ return new
StateValueColumnConfiguration(
+ columnIndex,
+
options.getOptional(stateNameOption)
+
.orElse(valueRowField.getName()),
+
options.getOptional(stateTypeOption)
+ .orElseGet(
+ () ->
+
inferStateType(
+
valueLogicalType)),
+
options.getOptional(mapKeyFormatOption)
+ .orElseGet(
+ () ->
+
inferStateMapKeyFormat(
+
valueRowField
+
.getName(),
+
valueLogicalType)),
+
options.getOptional(valueFormatOption)
+ .orElseGet(
+ () ->
+
inferStateValueFormat(
+
valueRowField
+
.getName(),
+
valueLogicalType)));
+ })
+ .collect(Collectors.toList()));
+ FactoryUtil.validateFactoryOptions(requiredOptions, optionalOptions,
options);
+
+ Set<String> consumedOptionKeys = new HashSet<>();
+ consumedOptionKeys.add(CONNECTOR.key());
+
requiredOptions.stream().map(ConfigOption::key).forEach(consumedOptionKeys::add);
+
optionalOptions.stream().map(ConfigOption::key).forEach(consumedOptionKeys::add);
+ FactoryUtil.validateUnconsumedKeys(
+ factoryIdentifier(), options.keySet(), consumedOptionKeys);
+
+ return new SavepointDynamicTableSource(
+ stateBackendType,
+ statePath,
+ operatorIdentifier,
+ keyFormat,
+ keyValueConfigProjections,
+ rowType);
+ }
+
+ private Tuple2<Integer, int[]>
createKeyValueProjections(ResolvedCatalogTable catalogTable) {
+ ResolvedSchema schema = catalogTable.getResolvedSchema();
+ if (schema.getPrimaryKey().isEmpty()) {
+ throw new ValidationException("Could not find the primary key in
the table schema.");
+ }
+
+ List<String> keyFields = schema.getPrimaryKey().get().getColumns();
+ if (keyFields.size() != 1) {
+ throw new ValidationException(
+ "Only a single primary key must be defined in the table
schema.");
+ }
+
+ DataType physicalDataType = schema.toPhysicalRowDataType();
+ int keyProjection = createKeyFormatProjection(physicalDataType,
keyFields.get(0));
+ int[] valueProjection = createValueFormatProjection(physicalDataType,
keyProjection);
+
+ return Tuple2.of(keyProjection, valueProjection);
+ }
+
+ private int createKeyFormatProjection(DataType physicalDataType, String
keyField) {
+ final LogicalType physicalType = physicalDataType.getLogicalType();
+ Preconditions.checkArgument(
+ physicalType.is(LogicalTypeRoot.ROW), "Row data type
expected.");
+ final List<String> physicalFields =
LogicalTypeChecks.getFieldNames(physicalType);
+ return physicalFields.indexOf(keyField);
+ }
+
+ private int[] createValueFormatProjection(DataType physicalDataType, int
keyProjection) {
+ final LogicalType physicalType = physicalDataType.getLogicalType();
+ Preconditions.checkArgument(
+ physicalType.is(LogicalTypeRoot.ROW), "Row data type
expected.");
+ final int physicalFieldCount =
LogicalTypeChecks.getFieldCount(physicalType);
+ final IntStream physicalFields = IntStream.range(0,
physicalFieldCount);
+
+ return physicalFields.filter(pos -> keyProjection != pos).toArray();
+ }
+
+ private SavepointConnectorOptions.StateType inferStateType(LogicalType
logicalType) {
+ switch (logicalType.getTypeRoot()) {
+ case ARRAY:
+ return SavepointConnectorOptions.StateType.LIST;
+
+ case MAP:
+ return SavepointConnectorOptions.StateType.MAP;
+
+ default:
+ return SavepointConnectorOptions.StateType.VALUE;
+ }
+ }
+
+ @Nullable
+ private String inferStateMapKeyFormat(String columnName, LogicalType
logicalType) {
+ return logicalType.is(LogicalTypeRoot.MAP)
+ ? inferStateValueFormat(columnName, ((MapType)
logicalType).getKeyType())
+ : null;
+ }
+
+ private String inferStateValueFormat(String columnName, LogicalType
logicalType) {
+ switch (logicalType.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ return String.class.getName();
+
+ case BOOLEAN:
+ return Boolean.class.getName();
+
+ case BINARY:
+ case VARBINARY:
+ return byte[].class.getName();
+
+ case DECIMAL:
+ return BigDecimal.class.getName();
+
+ case TINYINT:
+ return Byte.class.getName();
+
+ case SMALLINT:
+ return Short.class.getName();
+
+ case INTEGER:
+ return Integer.class.getName();
+
+ case BIGINT:
+ return Long.class.getName();
+
+ case FLOAT:
+ return Float.class.getName();
+
+ case DOUBLE:
+ return Double.class.getName();
+
+ case DATE:
+ return Integer.class.getName();
+
+ case INTERVAL_YEAR_MONTH:
+ case INTERVAL_DAY_TIME:
+ return Long.class.getName();
+
+ case ARRAY:
+ return inferStateValueFormat(
+ columnName, ((ArrayType)
logicalType).getElementType());
+
+ case MAP:
+ return inferStateValueFormat(columnName, ((MapType)
logicalType).getValueType());
+
+ case NULL:
+ return null;
+
+ case ROW:
+ case MULTISET:
+ case TIME_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ case DISTINCT_TYPE:
+ case STRUCTURED_TYPE:
+ case RAW:
+ case SYMBOL:
+ case UNRESOLVED:
+ default:
+ throw new UnsupportedOperationException(
+ String.format(
+ "Unable to infer state format for SQL type: %s
in column: %s. "
+ + "Please override the type with the
following config parameter: %s.%s.%s",
+ logicalType, columnName, FIELDS, columnName,
VALUE_FORMAT));
+ }
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return "savepoint";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(STATE_BACKEND_TYPE);
+ options.add(STATE_PATH);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+
+ // Either UID or hash
+ options.add(OPERATOR_UID);
+ options.add(OPERATOR_UID_HASH);
+
+ // Multiple values can be read so registering placeholders
+ options.add(STATE_NAME_PLACEHOLDER);
+ options.add(STATE_TYPE_PLACEHOLDER);
+ options.add(MAP_KEY_FORMAT_PLACEHOLDER);
+ options.add(VALUE_FORMAT_PLACEHOLDER);
+
+ return options;
+ }
+}
diff --git
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/StateValueColumnConfiguration.java
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/StateValueColumnConfiguration.java
new file mode 100644
index 00000000000..b572f1a564a
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/StateValueColumnConfiguration.java
@@ -0,0 +1,79 @@
+/*
+ * 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.state.StateDescriptor;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+
+/** Configuration for SQL state columns. */
+@SuppressWarnings("rawtypes")
+public class StateValueColumnConfiguration implements Serializable {
+ private final int columnIndex;
+ private final String stateName;
+ private final SavepointConnectorOptions.StateType stateType;
+ @Nullable private final String mapKeyFormat;
+ private final String valueFormat;
+ @Nullable private StateDescriptor stateDescriptor;
+
+ public StateValueColumnConfiguration(
+ int columnIndex,
+ final String stateName,
+ final SavepointConnectorOptions.StateType stateType,
+ @Nullable final String mapKeyFormat,
+ @Nullable final String valueFormat) {
+ this.columnIndex = columnIndex;
+ this.stateName = stateName;
+ this.stateType = stateType;
+ this.mapKeyFormat = mapKeyFormat;
+ this.valueFormat = valueFormat;
+ }
+
+ public int getColumnIndex() {
+ return columnIndex;
+ }
+
+ public String getStateName() {
+ return stateName;
+ }
+
+ public SavepointConnectorOptions.StateType getStateType() {
+ return stateType;
+ }
+
+ @Nullable
+ public String getMapKeyFormat() {
+ return mapKeyFormat;
+ }
+
+ public String getValueFormat() {
+ return valueFormat;
+ }
+
+ public void setStateDescriptor(StateDescriptor stateDescriptor) {
+ this.stateDescriptor = stateDescriptor;
+ }
+
+ @Nullable
+ public StateDescriptor getStateDescriptor() {
+ return stateDescriptor;
+ }
+}
diff --git
a/flink-libraries/flink-state-processing-api/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
b/flink-libraries/flink-state-processing-api/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 00000000000..730e50e185e
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.state.table.SavepointDynamicTableSourceFactory
diff --git
a/flink-libraries/flink-state-processing-api/src/test/java/com/example/state/writer/job/schema/PojoData.java
b/flink-libraries/flink-state-processing-api/src/test/java/com/example/state/writer/job/schema/PojoData.java
new file mode 100644
index 00000000000..74cc20d07bd
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/test/java/com/example/state/writer/job/schema/PojoData.java
@@ -0,0 +1,33 @@
+/*
+ * 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 com.example.state.writer.job.schema;
+
+/** Example POJO data to deserialize state. */
+public class PojoData {
+ private Long privateLong;
+ public Long publicLong;
+
+ public Long getPrivateLong() {
+ return privateLong;
+ }
+
+ public void setPrivateLong(Long privateLong) {
+ this.privateLong = privateLong;
+ }
+}
diff --git
a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointDynamicTableSourceTest.java
b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointDynamicTableSourceTest.java
new file mode 100644
index 00000000000..c31efeafbeb
--- /dev/null
+++
b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointDynamicTableSourceTest.java
@@ -0,0 +1,172 @@
+/*
+ * 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.RuntimeExecutionMode;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for the state SQL reader. */
+public class SavepointDynamicTableSourceTest {
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testReadKeyedState() throws Exception {
+ Configuration config = new Configuration();
+ config.set(RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(config);
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ final String sql =
+ "CREATE TABLE state_table (\n"
+ + " k bigint,\n"
+ + " KeyedPrimitiveValue bigint,\n"
+ + " KeyedPojoValue ROW<privateLong bigint, publicLong
bigint>,\n"
+ + " KeyedPrimitiveValueList ARRAY<bigint>,\n"
+ + " KeyedPrimitiveValueMap MAP<bigint, bigint>,\n"
+ + " PRIMARY KEY (k) NOT ENFORCED\n"
+ + ")\n"
+ + "with (\n"
+ + " 'connector' = 'savepoint',\n"
+ + " 'state.backend.type' = 'hashmap',\n"
+ + " 'state.path' =
'src/test/resources/table-state',\n"
+ + " 'operator.uid' = 'keyed-state-process-uid',\n"
+ + " 'fields.KeyedPojoValue.value-format' =
'com.example.state.writer.job.schema.PojoData'\n"
+ + ")";
+ tEnv.executeSql(sql);
+ Table table = tEnv.sqlQuery("SELECT * FROM state_table");
+ List<Row> result = tEnv.toDataStream(table).executeAndCollect(100);
+
+ assertThat(result.size()).isEqualTo(10);
+
+ // Check key
+ List<Long> keys =
+ result.stream().map(r -> (Long)
r.getField("k")).collect(Collectors.toList());
+ List<Long> expectedKeys = LongStream.range(0L,
10L).boxed().collect(Collectors.toList());
+ assertThat(keys).containsExactlyInAnyOrderElementsOf(expectedKeys);
+
+ // Check primitive value state
+ Set<Long> primitiveValues =
+ result.stream()
+ .map(r -> (Long) r.getField("KeyedPrimitiveValue"))
+ .collect(Collectors.toSet());
+ assertThat(primitiveValues.size()).isEqualTo(1);
+ assertThat(primitiveValues.iterator().next()).isEqualTo(1L);
+
+ // Check pojo value state
+ Set<Row> pojoValues =
+ result.stream()
+ .map(r -> (Row) r.getField("KeyedPojoValue"))
+ .collect(Collectors.toSet());
+ assertThat(pojoValues.size()).isEqualTo(1);
+ Row pojoData = pojoValues.iterator().next();
+ assertThat(pojoData.getField("publicLong")).isEqualTo(1L);
+ assertThat(pojoData.getField("privateLong")).isEqualTo(1L);
+
+ // Check list state
+ Set<Tuple2<Long, Long[]>> listValues =
+ result.stream()
+ .map(
+ r ->
+ Tuple2.of(
+ (Long) r.getField("k"),
+ (Long[])
r.getField("KeyedPrimitiveValueList")))
+ .flatMap(l -> Set.of(l).stream())
+ .collect(Collectors.toSet());
+ assertThat(listValues.size()).isEqualTo(10);
+ for (Tuple2<Long, Long[]> tuple2 : listValues) {
+ assertThat(tuple2.f0).isEqualTo(tuple2.f1[0]);
+ }
+
+ // Check map state
+ Set<Tuple2<Long, Map<Long, Long>>> mapValues =
+ result.stream()
+ .map(
+ r ->
+ Tuple2.of(
+ (Long) r.getField("k"),
+ (Map<Long, Long>)
+
r.getField("KeyedPrimitiveValueMap")))
+ .flatMap(l -> Set.of(l).stream())
+ .collect(Collectors.toSet());
+ assertThat(mapValues.size()).isEqualTo(10);
+ for (Tuple2<Long, Map<Long, Long>> tuple2 : mapValues) {
+ assertThat(tuple2.f1.size()).isEqualTo(1);
+ assertThat(tuple2.f0).isEqualTo(tuple2.f1.get(tuple2.f0));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("DataFlowIssue")
+ public void testReadKeyedStateWithNullValues() throws Exception {
+ Configuration config = new Configuration();
+ config.set(RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(config);
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ final String sql =
+ "CREATE TABLE state_table (\n"
+ + " k bigint,\n"
+ + " total ROW<privateLong bigint, publicLong
bigint>,\n"
+ + " PRIMARY KEY (k) NOT ENFORCED\n"
+ + ")\n"
+ + "with (\n"
+ + " 'connector' = 'savepoint',\n"
+ + " 'state.backend.type' = 'hashmap',\n"
+ + " 'state.path' =
'src/test/resources/table-state-nulls',\n"
+ + " 'operator.uid' =
'keyed-state-process-uid-null',\n"
+ + " 'fields.total.value-format' =
'com.example.state.writer.job.schema.PojoData'\n"
+ + ")";
+ tEnv.executeSql(sql);
+ Table table = tEnv.sqlQuery("SELECT * FROM state_table");
+ List<Row> result = tEnv.toDataStream(table).executeAndCollect(100);
+ assertThat(result.size()).isEqualTo(5);
+
+ List<Long> keys =
+ result.stream().map(row -> (Long)
row.getField("k")).collect(Collectors.toList());
+ assertThat(keys).containsExactlyInAnyOrder(1L, 2L, 3L, 4L, 5L);
+
+ // Check pojo value state
+ Map<Long, Row> pojoValues =
+ result.stream()
+ .collect(
+ Collectors.toMap(
+ v -> (Long) v.getField("k"),
+ v -> (Row) v.getField("total")));
+ assertThat(pojoValues.get(1L)).isEqualTo(Row.of(1L, 1L));
+ assertThat(pojoValues.get(2L)).isEqualTo(Row.of(null, null));
+ assertThat(pojoValues.get(3L)).isEqualTo(Row.of(null, null));
+ assertThat(pojoValues.get(4L)).isEqualTo(Row.of(4L, 4L));
+ assertThat(pojoValues.get(5L)).isEqualTo(Row.of(5L, 5L));
+ }
+}
diff --git
a/flink-libraries/flink-state-processing-api/src/test/resources/table-state-nulls/_metadata
b/flink-libraries/flink-state-processing-api/src/test/resources/table-state-nulls/_metadata
new file mode 100644
index 00000000000..4ac68dbf7bf
Binary files /dev/null and
b/flink-libraries/flink-state-processing-api/src/test/resources/table-state-nulls/_metadata
differ
diff --git
a/flink-libraries/flink-state-processing-api/src/test/resources/table-state/_metadata
b/flink-libraries/flink-state-processing-api/src/test/resources/table-state/_metadata
new file mode 100644
index 00000000000..dc9d5acbbd2
Binary files /dev/null and
b/flink-libraries/flink-state-processing-api/src/test/resources/table-state/_metadata
differ
