[FLINK-37407][state] Add savepoint metadata SQL built-in process function #26393

gaborgsomogyi · 2025-04-03T07:39:47Z

What is the purpose of the change

Savepoint metadata is useful information from user perspective but it's not yet available from SQL perspective. In this PR I've added support for this, please see FLIP512 for further details.

I've also added some tiny name rationalization in comments.

Brief change log

Added savepoint metadata SQL built-in process function.

Verifying this change

New automated test.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs

gaborgsomogyi · 2025-04-03T07:41:06Z

docs/content/docs/libs/state_processor_api.md

+| operator-uid-hash                        | STRING NOT NULL | Operator UID hash.                                                               |
+| operator-parallelism                     | INT NOT NULL    | Parallelism of the operator.                                                     |
+| operator-max-parallelism                 | INT NOT NULL    | Maximum parallelism of the operator.                                             |
+| operator-subtask-state-count             | INT NOT NULL    | Number of operator subtask states.                                               |


operator-subtask-state-count is an addition compared to the FLIP. I thought it's valuable information but if somebody has objection we can remove it.

It looks like a good addition to me

flinkbot · 2025-04-03T07:45:51Z

CI report:

a32dc3d Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

davidradl · 2025-04-03T10:07:31Z

docs/content/docs/libs/state_processor_api.md

@@ -515,6 +515,28 @@ Before you interrogate state using the table API, make sure to review our [Flink

 IMPORTANT NOTE: State Table API only supports keyed state.

+### Metadata
+
+The SQL connector allows users to read metadata of savepoints in the following way:


nit:
read metadata -> read the metadata
This is not a SQL connector - it is a new table function.

davidradl · 2025-04-03T10:10:48Z

docs/content/docs/libs/state_processor_api.md

+The SQL connector allows users to read metadata of savepoints in the following way:
+```SQL
+LOAD MODULE state;
+SELECT * FROM savepoint_metadata('/root/dir/of/checkpoint-data/chk-1');


The flip talks of checkpoints and savepoint state, can we use this funciton to access checkpoint state as well? Currently the function is savepoint_metadata; if it covers both can we function change the name so it applies to both savepoints and checkpoints - maybe state_table_metadata?

If checkpoints are not included - how does it error if we point this function at the checkpoint file? Maybe a unit test for that would be good.

The flip talks of checkpoints and savepoint state, can we use this funciton to access checkpoint state as well?

Yes. We can't name things in both ways. Initially we've agreed to call the whole state processor API things as savepoint... and the only exception here is the module because it's a reserved keyword in Flink SQL. If we go on this road, then the whole state processor API including internals must be renamed which I don't support.

davidradl · 2025-04-03T10:14:22Z

docs/content/docs/libs/state_processor_api.md

+SELECT * FROM savepoint_metadata('/root/dir/of/checkpoint-data/chk-1');
+```
+
+The table have the following fixed schema:


nit:
The table have the following fixed schema:
->
The new table function creates a table with the following fixed schema:

davidradl · 2025-04-03T10:16:54Z

...ate-processing-api/src/main/java/org/apache/flink/state/table/SavepointConnectorOptions.java

@@ -26,7 +26,7 @@

 import static org.apache.flink.configuration.description.TextElement.code;

-/** Options for the state connector. */
+/** Options for the savepoint connector. */


nit: this is not really a connector as far as I can see.

This change is unrelated to the meta feature, just renamed everything to savepoint as we've agreed on. This is used by the connector...

davidradl · 2025-04-03T10:21:51Z

...rocessing-api/src/main/java/org/apache/flink/state/table/SavepointMetadataTableFunction.java

+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.types.Row;
+
+@Internal


is this internal -as it is driven directly from SQL?

Similar classes are having such annotation.

davidradl · 2025-04-03T10:23:33Z

...link-state-processing-api/src/main/java/org/apache/flink/state/table/module/StateModule.java

+
+import static org.apache.flink.table.functions.FunctionKind.TABLE;
+
+/** Module of state in Flink. */


we should mention savepoint if this is only savepoint

Not only. As mentioned before everything is called savepoint except the module because that's a reserved keyword.

davidradl · 2025-04-03T10:30:43Z

docs/content/docs/libs/state_processor_api.md

+
+The SQL connector allows users to read metadata of savepoints in the following way:
+```SQL
+LOAD MODULE state;


nit: if this is savepoints only ; should we call it savepoint-state. Currently the implementation seems to be tied to a fixed savepoint schema. savepoint-state would be more descriptive and better scoped - allowing us to provide a more generic state module later - if we want to.

Seems like all the rest of the questions related is savepoint only?. My answer to all of them is no. We've agreed on this naming in the discussion.

davidradl · 2025-04-03T10:33:51Z

...link-state-processing-api/src/main/java/org/apache/flink/state/table/module/StateModule.java

+@Experimental
+public class StateModule implements Module {
+
+    public static final String IDENTIFIER = "state";


I think the identified should mention the word savepoint maybe "savepoint-state"

davidradl · 2025-04-03T10:36:19Z

...link-state-processing-api/src/main/java/org/apache/flink/state/table/module/StateModule.java

+
+/** Module of state in Flink. */
+@Experimental
+public class StateModule implements Module {


I suggest we rename SavepointStateModule so it is more descriptive
I wonder if we should refactor CoreModule so we can extend, as the only difference seems to be the List<BuiltInFunctionDefinition> definitions

gaborgsomogyi · 2025-04-03T17:23:57Z

cc @Zakelly

Zakelly · 2025-04-07T04:13:17Z

...rocessing-api/src/main/java/org/apache/flink/state/table/SavepointMetadataTableFunction.java

+                row.setField("operator-parallelism", operatorState.getParallelism());
+                row.setField("operator-max-parallelism", operatorState.getMaxParallelism());
+                row.setField("operator-subtask-state-count", operatorState.getStates().size());


Ah, is operator-parallelism and operator-subtask-state-count providing the same value? It seems so.

No, for example broadcast source has operator-parallelism = 1 but operator-subtask-state-count = 0.

Thus I suggest adding a more detailed description for operator-subtask-state-count in the doc. It represents the state partition count divided by the operator's parallelism and might be 0 if the state is not partitioned. That's my understanding, correct me if I'm wrong.

Yeah, this is a hidden gem and worth to add some more info. Your understanding is correct so I would add your sentence as-is.

Zakelly · 2025-04-07T04:25:14Z

...link-state-processing-api/src/main/java/org/apache/flink/state/table/module/StateModule.java

+    private final Set<String> functionNamesWithInternal;
+    private final Set<String> functionNamesWithoutInternal;
+
+    public StateModule() {


We could make constructor private since we are using singleton

Zakelly

Thanks for the update. LGTM, only one minor suggestion

Zakelly · 2025-04-07T08:03:58Z

...rocessing-api/src/main/java/org/apache/flink/state/table/SavepointMetadataTableFunction.java

+                row.setField("operator-parallelism", operatorState.getParallelism());
+                row.setField("operator-max-parallelism", operatorState.getMaxParallelism());
+                row.setField("operator-subtask-state-count", operatorState.getStates().size());


Thus I suggest adding a more detailed description for operator-subtask-state-count in the doc. It represents the state partition count divided by the operator's parallelism and might be 0 if the state is not partitioned. That's my understanding, correct me if I'm wrong.

…tion

gaborgsomogyi commented Apr 3, 2025

View reviewed changes

flinkbot added the component=API/StateProcessor label Apr 3, 2025

davidradl reviewed Apr 3, 2025

View reviewed changes

gaborgsomogyi force-pushed the FLINK-37407 branch from 9396b6f to f8320ef Compare April 3, 2025 13:15

Zakelly reviewed Apr 7, 2025

View reviewed changes

gaborgsomogyi force-pushed the FLINK-37407 branch from f8320ef to a224652 Compare April 7, 2025 06:19

Zakelly approved these changes Apr 7, 2025

View reviewed changes

gaborgsomogyi force-pushed the FLINK-37407 branch from a224652 to 3521598 Compare April 7, 2025 08:15

Zakelly approved these changes Apr 7, 2025

View reviewed changes

[FLINK-37407][state] Add savepoint metadata SQL built-in process func…

a32dc3d

…tion

gaborgsomogyi force-pushed the FLINK-37407 branch from 3521598 to a32dc3d Compare April 7, 2025 12:09

gaborgsomogyi merged commit ae09a0d into apache:master Apr 7, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-37407][state] Add savepoint metadata SQL built-in process function #26393

[FLINK-37407][state] Add savepoint metadata SQL built-in process function #26393

gaborgsomogyi commented Apr 3, 2025

gaborgsomogyi Apr 3, 2025

davidradl Apr 3, 2025

flinkbot commented Apr 3, 2025 •

edited

Loading

davidradl Apr 3, 2025 •

edited

Loading

gaborgsomogyi Apr 3, 2025

davidradl Apr 3, 2025 •

edited

Loading

gaborgsomogyi Apr 3, 2025

davidradl Apr 3, 2025 •

edited

Loading

gaborgsomogyi Apr 3, 2025

davidradl Apr 3, 2025 •

edited

Loading

gaborgsomogyi Apr 3, 2025

davidradl Apr 3, 2025

gaborgsomogyi Apr 3, 2025

davidradl Apr 3, 2025

gaborgsomogyi Apr 3, 2025

davidradl Apr 3, 2025

gaborgsomogyi Apr 3, 2025

davidradl Apr 3, 2025

davidradl Apr 3, 2025 •

edited

Loading

gaborgsomogyi commented Apr 3, 2025

Zakelly Apr 7, 2025

gaborgsomogyi Apr 7, 2025 •

edited

Loading

Zakelly Apr 7, 2025

gaborgsomogyi Apr 7, 2025

Zakelly Apr 7, 2025

gaborgsomogyi Apr 7, 2025

Zakelly left a comment

Zakelly Apr 7, 2025


		import static org.apache.flink.table.functions.FunctionKind.TABLE;

		/** Module of state in Flink. */

[FLINK-37407][state] Add savepoint metadata SQL built-in process function #26393

[FLINK-37407][state] Add savepoint metadata SQL built-in process function #26393

Conversation

gaborgsomogyi commented Apr 3, 2025

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flinkbot commented Apr 3, 2025 • edited Loading

CI report:

davidradl Apr 3, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidradl Apr 3, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidradl Apr 3, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidradl Apr 3, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidradl Apr 3, 2025 • edited Loading

Choose a reason for hiding this comment

gaborgsomogyi commented Apr 3, 2025

Choose a reason for hiding this comment

gaborgsomogyi Apr 7, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Zakelly left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flinkbot commented Apr 3, 2025 •

edited

Loading

davidradl Apr 3, 2025 •

edited

Loading

davidradl Apr 3, 2025 •

edited

Loading

davidradl Apr 3, 2025 •

edited

Loading

davidradl Apr 3, 2025 •

edited

Loading

davidradl Apr 3, 2025 •

edited

Loading

gaborgsomogyi Apr 7, 2025 •

edited

Loading