[FLINK-32053][table-planner] Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan #22593

LadyForest · 2023-05-16T12:26:23Z

What is the purpose of the change

This PR is a subtask of FLIP-292 to enable operator-level state TTL configuration via CompiledPlan.

Brief change log

Introduce StateMetadata to all ExecNodes that translate to stateful operators, and changes the way how #translateToPlanInternal get the state retention time.

The affected ExecNode list

    StreamExecChangelogNormalize
    StreamExecDeduplicate
    StreamExecGlobalGroupAggregate
    StreamExecGroupAggregate
    StreamExecIncrementalGroupAggregate
    StreamExecJoin
    StreamExecLimit
    StreamExecLookupJoin
    StreamExecRank
    StreamExecSink
    StreamExecSortLimit

Verifying this change

Since we have upgraded some ExecNodes to version 2, we have to test the following 3 parts:

The plan serialized using version 1 can be deserialized using the current version. This can be verified by TransformationsTest#testUidFlink1_15
The plan with the current version SerDe work as expected. This can be verified by all tests under package org.apache.flink.table.planner.plan.nodes.exec.stream
The way by modifying the JSON content to change state TTL works as expected. This can be verified by ITCase ConfigureOperatorLevelStateTtlJsonITCase

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 introduces a new feature? yes
If yes, how is the feature documented? FLINK-31957

flinkbot · 2023-05-16T12:32:38Z

CI report:

45589a9 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

godfreyhe · 2023-05-19T02:23:06Z

...java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecChangelogNormalize.java

@@ -71,20 +75,36 @@
        producedTransformations = StreamExecChangelogNormalize.CHANGELOG_NORMALIZE_TRANSFORMATION,
        minPlanVersion = FlinkVersion.v1_15,
        minStateVersion = FlinkVersion.v1_15)
+@ExecNodeMetadata(
+        name = "stream-exec-changelog-normalize",
+        version = 2,


Do we really need to introduce a new version ? Because the topology and state layout do not change,

Do we really need to introduce a new version ? Because the topology and state layout do not change,

I also hesitated for a long time on this point. Introducing StateMetadata did not actually change the state layout, but the FLIP-190 design document provides an example of upgrading the exec node version after adding a member variable.

The JSON node of ExecNode A gets an additional property in Flink 1.16:

Before:

{
some-prop: 42
}
After:

{
some-prop: 42,
some-flag: false
}

Every plan change will increase the ExecNode version. Since the ExecNode supports both plan layouts, adding an additional annotation is enough.

@ExecNodeMetadata(name=A, version=1, minPlanVersion=1.15, minStateVersion=1.15)
@ExecNodeMetadata(name=A, version=2, minPlanVersion=1.16, minStateVersion=1.15)
class MyExecNode extends ExecNode {...}
New plans will use the new ExecNode version 2.

After reading the doc of the method minPlanVersion

Needs to be updated when the JSON for the {@link ExecNode} changes: e.g. after adding an
attribute to the JSON spec of the ExecNode.

I think we need introduce a new version since the json for the node changes.

@godfreyhe @luoyuxia @LadyForest If the testing infrastructure would be already in place, I would agree with increasing the version. But we are not testing different versions of ExecNode yet. So increasing the annotation has no effect.

I just noticed that you bumped the ExecNode version of a couple of ExecNodes for the new state metadata properties in CompiledPlan. Your change actually allows having null for the stateMetadataList , I'm wondering if we should not increase the version for 1.18. We can stay at the old ExecNode version as 1.18 is able to consume 1.17 plans and state.
The reason why noticed that is due to FLINK-32613 where I added an option to the wrong annotation during rebase and no test failed.

Hi @twalthr, during the implementation of FLIP-292, I did notice that the current testing for compiled plans is not comprehensive enough, such as FLINK-31884, FLINK-31917, and FLINK-32219. FLIP-292 added ExecNodeVersionUpgradeSerdeTest and TransformationsTest#testUidFlink1_15 to test the deserialization of old plans using the new version, but it may not be sufficient. After reconsideration, I think we can keep the original exec node version until we have a more comprehensive testing framework. I will open a PR and hope you can help review the code. Thank you.

Hi @LadyForest, sounds good to me. Thanks for reverting the version bump. Btw FLINK-31917 seems pretty serious. Are we still able to deserialize 1.15 literals with this change? Also the SARG values cannot be null anymore? I guess that should be fine. Ping me in Slack, I can help with a review.

Are we still able to deserialize 1.15 literals with this change?

I suppose so. I adapted the test TransformationsTest#testUidFlink1_15 to verify the change. The test deserialized the serialized plan using Flink v1.15, where numbers are serialized as string literals.

Also the SARG values cannot be null anymore?

Sorry, I'm a bit confused on this point. Did you find any incompatibilities? AFAIK, the changes made on RexNodeJsonSerializer does not aim to change the processing logic for SARG. You can check RexNodeJsonSerdeTest and SargJsonPlanITCase for more details.

Hi @twalthr, it appears that modifying an already approved FLIP requires a new vote. Considering that feature cut for 1.18 is only a few days away, do you have any suggestions for a quick fix?

...nner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java

luoyuxia

@LadyForest Thanks for contribution. I left some comments. PTAL.

...able-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/StateMetadata.java

...rc/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java

luoyuxia · 2023-06-02T09:38:45Z

...java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecChangelogNormalize.java

@@ -71,20 +75,36 @@
        producedTransformations = StreamExecChangelogNormalize.CHANGELOG_NORMALIZE_TRANSFORMATION,
        minPlanVersion = FlinkVersion.v1_15,
        minStateVersion = FlinkVersion.v1_15)
+@ExecNodeMetadata(
+        name = "stream-exec-changelog-normalize",
+        version = 2,


After reading the doc of the method minPlanVersion

Needs to be updated when the JSON for the {@link ExecNode} changes: e.g. after adding an
attribute to the JSON spec of the ExecNode.

I think we need introduce a new version since the json for the node changes.

...c/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java

...lanner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/TransformationsTest.java

...flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/JsonPlanTestBase.java

luoyuxia · 2023-06-02T12:08:41Z

...he/flink/table/planner/runtime/stream/jsonplan/ConfigureOperatorLevelStateTtlJsonITCase.java

+                                JsonNode target = JsonTestUtils.readFromString(json);
+                                JsonTestUtils.setExecNodeStateMetadata(
+                                        target, "stream-exec-deduplicate", 0, 6000L);
+                                JsonTestUtils.setExecNodeStateMetadata(


I'm wondering whether it'll be better to change it 8000L.
The reason is that if we remove this code line, the test can still passs so that we can't make sure set state for stream-exec-group-aggregate also make difference.

Commenting out this line means that the TTL of the deduplication node will take the default value of 0, which means that the status will never expire. However, the actual effect is the same as setting a TTL of 6 or 8 seconds.

I think this case demonstrates the significance of supporting operator-level TTL configuration. That is to say, as long as we ensure that the data calculation results remain unchanged, we do not need to unnecessarily extend the state TTL for the sake of a long-period state of a certain operator.

Let me add more description to the case.

…upport configure operator-level state TTL via CompiledPlan

luoyuxia

LGTM

LadyForest force-pushed the FLINK-32053 branch from 1351fed to 9fe8b56 Compare May 16, 2023 15:52

godfreyhe reviewed May 19, 2023

View reviewed changes

LadyForest requested a review from godfreyhe May 23, 2023 06:32

luoyuxia reviewed Jun 2, 2023

View reviewed changes

LadyForest added 2 commits June 3, 2023 23:13

[FLINK-32053][table-planner] Introduce StateMetadata to ExecNode to s…

c799676

…upport configure operator-level state TTL via CompiledPlan

address comments

da5ff67

LadyForest force-pushed the FLINK-32053 branch from 58356b5 to 5c55e07 Compare June 3, 2023 15:13

LadyForest requested a review from luoyuxia June 4, 2023 10:22

address comments

e6a7587

LadyForest force-pushed the FLINK-32053 branch from 5c55e07 to e6a7587 Compare June 5, 2023 03:19

address comments

45589a9

luoyuxia approved these changes Jun 5, 2023

View reviewed changes

luoyuxia merged commit 69def87 into apache:master Jun 5, 2023

LadyForest mentioned this pull request Jul 23, 2023

[FLINK-32657][table-planner] Revert upgrading ExecNode versions for StateMetadata #23054

Closed

flinkbot added the component=TableSQL/Planner label Apr 4, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-32053][table-planner] Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan #22593

[FLINK-32053][table-planner] Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan #22593

LadyForest commented May 16, 2023

flinkbot commented May 16, 2023 •

edited

godfreyhe May 19, 2023

LadyForest May 23, 2023

luoyuxia Jun 2, 2023

twalthr Jul 19, 2023 •

edited

LadyForest Jul 20, 2023

twalthr Jul 20, 2023

LadyForest Jul 20, 2023

LadyForest Jul 21, 2023

luoyuxia left a comment

luoyuxia Jun 2, 2023

luoyuxia Jun 2, 2023

LadyForest Jun 3, 2023

luoyuxia left a comment

[FLINK-32053][table-planner] Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan #22593

[FLINK-32053][table-planner] Introduce StateMetadata to ExecNode to support configure operator-level state TTL via CompiledPlan #22593

Conversation

LadyForest commented May 16, 2023

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

flinkbot commented May 16, 2023 • edited

CI report:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

twalthr Jul 19, 2023 • edited

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

luoyuxia left a 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

luoyuxia left a comment

Choose a reason for hiding this comment

flinkbot commented May 16, 2023 •

edited

twalthr Jul 19, 2023 •

edited