You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "LadyForest (via GitHub)" <gi...@apache.org> on 2023/04/27 06:39:24 UTC
[GitHub] [flink] LadyForest opened a new pull request, #22492: [FLINK-31950][table-planner] Introduce StateMetadata and support operator-level state TTL configuration via Compiled Plan
LadyForest opened a new pull request, #22492:
URL: https://github.com/apache/flink/pull/22492
## What is the purpose of the change
This is the first subtask of FLINK-31791 for FLIP-288.
## Brief change log
This PR contains 3 commits
- The 1st commit introduces `StateMetadata`, which describes the operator's state info, alongside with`StateMetadataJsonSerializer` and `StateMetadataJsonDeserializer`.
- The 2nd and 3rd commit are wait for FLINK-31917 and FLINK-31884 to be merged.
## Verifying this change
This change added tests and can be verified as follows:
- The 1st commit adds `StateMetadataTest` to verify the JsonSerde works as expected.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #22492: [FLINK-31950][table-planner] Introduce StateMetadata and StateMetadataJson SerDe
Posted by "godfreyhe (via GitHub)" <gi...@apache.org>.
godfreyhe commented on code in PR #22492:
URL: https://github.com/apache/flink/pull/22492#discussion_r1193258928
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/StateMetadata.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.TimeUtils;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * It is used to describe the state metadata of a stateful operator, which is
+ * serialized/deserialized into/from those {@link
+ * org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecNode}s that can generate stateful
+ * operators. For ExecNodes that generates {@link
+ * org.apache.flink.streaming.api.operators.TwoInputStreamOperator} or {@link
+ * org.apache.flink.streaming.api.operators.MultipleInputStreamOperator}, there will be multiple
+ * metadata describing information about each input's state.
+ *
+ * <p>The metadata describes the following attributes.
+ *
+ * <ul>
+ * <li>{@code stateIndex}: annotates the state is from the i-th input, index based on zero
+ * <li>{@code ttl}: annotates the state retention time for the i-th input's state, the time unit
+ * is ms.
+ * <li>{@code name}: annotates the state description, such as deduplicate-state, join-left-state.
+ * </ul>
+ */
+@Internal
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class StateMetadata {
+ public static final String FIELD_NAME_STATE_INDEX = "index";
+ public static final String FIELD_NAME_STATE_TTL = "ttl";
+ public static final String FIELD_NAME_STATE_NAME = "name";
+
+ @JsonProperty(value = FIELD_NAME_STATE_INDEX, index = 0)
+ private final int stateIndex;
+
+ @JsonProperty(value = FIELD_NAME_STATE_TTL, index = 1)
+ private final Duration stateTtl;
+
+ @JsonProperty(value = FIELD_NAME_STATE_NAME, index = 2)
+ private final String stateName;
+
+ @JsonCreator
+ public StateMetadata(
+ @JsonProperty(FIELD_NAME_STATE_INDEX) int stateIndex,
+ @JsonProperty(FIELD_NAME_STATE_TTL) String stateTtl,
+ @JsonProperty(FIELD_NAME_STATE_NAME) String stateName) {
+ this(
+ stateIndex,
+ TimeUtils.parseDuration(
+ Preconditions.checkNotNull(stateTtl, "state ttl should not be null")),
+ Preconditions.checkNotNull(stateName, "state name should not be null"));
+ }
+
+ public StateMetadata(int stateIndex, @Nonnull Duration stateTtl, @Nonnull String stateName) {
Review Comment:
nit: `@Nonnull` will check null value when running, you can improve it in the following prs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on pull request #22492: [FLINK-31950][table-planner] Introduce StateMetadata and support operator-level state TTL configuration via Compiled Plan
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on PR #22492:
URL: https://github.com/apache/flink/pull/22492#issuecomment-1525542072
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #22492: [FLINK-31950][table-planner] Introduce StateMetadata and support operator-level state TTL configuration via Compiled Plan
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22492:
URL: https://github.com/apache/flink/pull/22492#issuecomment-1524870395
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4da69ac609ceda1b471fc3c15d797217d2f4dd2e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4da69ac609ceda1b471fc3c15d797217d2f4dd2e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4da69ac609ceda1b471fc3c15d797217d2f4dd2e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #22492: [FLINK-31950][table-planner] Introduce StateMetadata and StateMetadataJson SerDe
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #22492:
URL: https://github.com/apache/flink/pull/22492#discussion_r1190908037
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/StateMetadata.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * It is used to describe the state metadata of a stateful operator, which is
+ * serialized/deserialized into/from those {@link
+ * org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecNode}s that can generate stateful
+ * operators. For ExecNodes that generates {@link
+ * org.apache.flink.streaming.api.operators.TwoInputStreamOperator} or {@link
+ * org.apache.flink.streaming.api.operators.MultipleInputStreamOperator}, there will be multiple
+ * metadata describing information about each input's state.
+ *
+ * <p>The metadata describes the following attributes.
+ *
+ * <ul>
+ * <li>{@code stateIndex}: annotates the state is from the i-th input, index based on zero
+ * <li>{@code ttl}: annotates the state retention time for the i-th input's state, the time unit
+ * is ms.
+ * <li>{@code name}: annotates the state description, such as deduplicate-state, join-left-state.
+ * </ul>
+ */
+@Internal
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class StateMetadata {
+ public static final String FIELD_NAME_STATE_INDEX = "index";
+ public static final String FIELD_NAME_STATE_TTL = "ttl";
+ public static final String FIELD_NAME_STATE_NAME = "name";
+
+ @JsonProperty(value = FIELD_NAME_STATE_INDEX, index = 0)
+ private final int stateIndex;
+
+ @JsonProperty(value = FIELD_NAME_STATE_TTL, index = 1)
+ private final long stateTtl;
+
+ @JsonProperty(value = FIELD_NAME_STATE_NAME, index = 2)
+ private final String stateName;
+
+ @JsonCreator
+ public StateMetadata(
+ @JsonProperty(FIELD_NAME_STATE_INDEX) int stateIndex,
+ @JsonProperty(FIELD_NAME_STATE_TTL) long stateTtl,
+ @JsonProperty(FIELD_NAME_STATE_NAME) String stateName) {
+ this.stateIndex = stateIndex;
+ this.stateTtl = stateTtl;
+ this.stateName = stateName;
+ }
+
+ public int getStateIndex() {
+ return stateIndex;
+ }
+
+ public long getStateTtl() {
+ return stateTtl;
+ }
+
+ public String getStateName() {
+ return stateName;
+ }
+
+ public static List<StateMetadata> getOneInputOperatorDefaultMeta(
+ ReadableConfig tableConfig, String stateName) {
+ long time = tableConfig.get(ExecutionConfigOptions.IDLE_STATE_RETENTION).toMillis();
Review Comment:
Remove this
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/StateMetadata.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * It is used to describe the state metadata of a stateful operator, which is
+ * serialized/deserialized into/from those {@link
+ * org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecNode}s that can generate stateful
+ * operators. For ExecNodes that generates {@link
+ * org.apache.flink.streaming.api.operators.TwoInputStreamOperator} or {@link
+ * org.apache.flink.streaming.api.operators.MultipleInputStreamOperator}, there will be multiple
+ * metadata describing information about each input's state.
+ *
+ * <p>The metadata describes the following attributes.
+ *
+ * <ul>
+ * <li>{@code stateIndex}: annotates the state is from the i-th input, index based on zero
+ * <li>{@code ttl}: annotates the state retention time for the i-th input's state, the time unit
+ * is ms.
+ * <li>{@code name}: annotates the state description, such as deduplicate-state, join-left-state.
+ * </ul>
+ */
+@Internal
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class StateMetadata {
+ public static final String FIELD_NAME_STATE_INDEX = "index";
+ public static final String FIELD_NAME_STATE_TTL = "ttl";
+ public static final String FIELD_NAME_STATE_NAME = "name";
+
+ @JsonProperty(value = FIELD_NAME_STATE_INDEX, index = 0)
+ private final int stateIndex;
+
+ @JsonProperty(value = FIELD_NAME_STATE_TTL, index = 1)
+ private final long stateTtl;
+
+ @JsonProperty(value = FIELD_NAME_STATE_NAME, index = 2)
+ private final String stateName;
+
+ @JsonCreator
+ public StateMetadata(
+ @JsonProperty(FIELD_NAME_STATE_INDEX) int stateIndex,
+ @JsonProperty(FIELD_NAME_STATE_TTL) long stateTtl,
+ @JsonProperty(FIELD_NAME_STATE_NAME) String stateName) {
+ this.stateIndex = stateIndex;
+ this.stateTtl = stateTtl;
+ this.stateName = stateName;
+ }
+
+ public int getStateIndex() {
+ return stateIndex;
+ }
+
+ public long getStateTtl() {
+ return stateTtl;
+ }
+
+ public String getStateName() {
+ return stateName;
+ }
+
+ public static List<StateMetadata> getOneInputOperatorDefaultMeta(
+ ReadableConfig tableConfig, String stateName) {
+ long time = tableConfig.get(ExecutionConfigOptions.IDLE_STATE_RETENTION).toMillis();
+ return Collections.singletonList(
+ new StateMetadata(
+ 0,
+ tableConfig.get(ExecutionConfigOptions.IDLE_STATE_RETENTION).toMillis(),
+ stateName));
+ }
+
+ public static List<StateMetadata> getMultiInputOperatorDefaultMeta(
+ ReadableConfig tableConfig, String... stateNameList) {
+ long stateRetentionTime =
+ tableConfig.get(ExecutionConfigOptions.IDLE_STATE_RETENTION).toMillis();
+ return IntStream.range(0, stateNameList.length)
+ .boxed()
+ .map(i -> new StateMetadata(i, stateRetentionTime, stateNameList[i]))
+ .collect(Collectors.toList());
+ }
+
+ public static long getStateTtlForOneInputOperator(
+ ExecNodeConfig config, @Nullable List<StateMetadata> stateMetadataList) {
+ return getStateTtlForMultiInputOperator(config, 1, stateMetadataList).get(0);
+ }
+
+ public static List<Long> getStateTtlForMultiInputOperator(
+ ExecNodeConfig config,
+ int inputNumOfOperator,
+ @Nullable List<StateMetadata> stateMetadataList) {
+ // for backward compatibility
+ if (CollectionUtil.isNullOrEmpty(stateMetadataList)) {
+ return Stream.generate(config::getStateRetentionTime)
+ .limit(inputNumOfOperator)
+ .collect(Collectors.toList());
+ }
+ // in case malformed json plan
+ validateStateMetadata(inputNumOfOperator, stateMetadataList);
+ return stateMetadataList.stream()
+ .sorted(Comparator.comparing(StateMetadata::getStateIndex))
+ .map(StateMetadata::getStateTtl)
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof StateMetadata)) {
+ return false;
+ }
+ StateMetadata that = (StateMetadata) o;
+ return stateIndex == that.stateIndex
+ && stateTtl == that.stateTtl
+ && stateName.equals(that.stateName);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(stateIndex, stateTtl, stateName);
+ }
+
+ /**
+ * Validate deserialized state metadata from json content of {@link
+ * org.apache.flink.table.api.CompiledPlan}.
+ *
+ * @param inputNumOfOperator the input number of the stateful operator that the exec node to
+ * translate to.
+ * @param stateMetadataList the deserialized state metadata list.
+ */
+ private static void validateStateMetadata(
+ int inputNumOfOperator, List<StateMetadata> stateMetadataList) {
+
+ // the state metadata list size should be equal to the input number of the operator
+ Preconditions.checkArgument(
+ inputNumOfOperator == stateMetadataList.size(),
+ String.format(
+ "The compiled plan contains inconsistent state metadata configuration.\n"
+ + "Received %s state meta for a %sInputStreamOperator.",
+ stateMetadataList.size(),
+ inputNumOfOperator > 2
+ ? "Multiple"
+ : inputNumOfOperator == 2 ? "Two" : "Single"));
Review Comment:
Nit: SingleInput -> OneInput
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #22492: [FLINK-31950][table-planner] Introduce StateMetadata and StateMetadataJson SerDe
Posted by "godfreyhe (via GitHub)" <gi...@apache.org>.
godfreyhe commented on code in PR #22492:
URL: https://github.com/apache/flink/pull/22492#discussion_r1192099873
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeConfig.java:
##########
@@ -55,7 +55,7 @@ private ExecNodeConfig(
this.isCompiled = isCompiled;
}
- static ExecNodeConfig of(
+ public static ExecNodeConfig of(
Review Comment:
The reason changing the access level to public is this method can be used in testing ? If that, please add `VisibleForTesting` for the method
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/StateMetadataJsonSerializer.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.serde;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.exec.StateMetadata;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer;
+
+import java.io.IOException;
+
+import static org.apache.flink.table.planner.plan.nodes.exec.StateMetadata.FIELD_NAME_STATE_INDEX;
+import static org.apache.flink.table.planner.plan.nodes.exec.StateMetadata.FIELD_NAME_STATE_NAME;
+import static org.apache.flink.table.planner.plan.nodes.exec.StateMetadata.FIELD_NAME_STATE_TTL;
+
+/**
+ * JSON serializer for {@link StateMetadata}.
+ *
+ * @see StateMetadataJsonDeserializer for the reverse operation.
+ */
+@Internal
+public class StateMetadataJsonSerializer extends StdSerializer<StateMetadata> {
+
+ private static final long serialVersionUID = 1L;
+
+ StateMetadataJsonSerializer() {
+ super(StateMetadata.class);
+ }
+
+ @Override
+ public void serialize(
+ StateMetadata stateMetadata,
+ JsonGenerator jsonGenerator,
+ SerializerProvider serializerProvider)
+ throws IOException {
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeNumberField(FIELD_NAME_STATE_INDEX, stateMetadata.getStateIndex());
+ jsonGenerator.writeStringField(FIELD_NAME_STATE_TTL, stateMetadata.getStateTtl() + " ms");
Review Comment:
How about we use `Duration` type to avoid StateMetadataJsonSerializer/StateMetadataJsonDeserializer ?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/StateMetadata.java:
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * It is used to describe the state metadata of a stateful operator, which is
+ * serialized/deserialized into/from those {@link
+ * org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecNode}s that can generate stateful
+ * operators. For ExecNodes that generates {@link
+ * org.apache.flink.streaming.api.operators.TwoInputStreamOperator} or {@link
+ * org.apache.flink.streaming.api.operators.MultipleInputStreamOperator}, there will be multiple
+ * metadata describing information about each input's state.
+ *
+ * <p>The metadata describes the following attributes.
+ *
+ * <ul>
+ * <li>{@code stateIndex}: annotates the state is from the i-th input, index based on zero
+ * <li>{@code ttl}: annotates the state retention time for the i-th input's state, the time unit
+ * is ms.
+ * <li>{@code name}: annotates the state description, such as deduplicate-state, join-left-state.
+ * </ul>
+ */
+@Internal
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class StateMetadata {
+ public static final String FIELD_NAME_STATE_INDEX = "index";
+ public static final String FIELD_NAME_STATE_TTL = "ttl";
+ public static final String FIELD_NAME_STATE_NAME = "name";
+
+ @JsonProperty(value = FIELD_NAME_STATE_INDEX, index = 0)
+ private final int stateIndex;
+
+ @JsonProperty(value = FIELD_NAME_STATE_TTL, index = 1)
+ private final long stateTtl;
+
+ @JsonProperty(value = FIELD_NAME_STATE_NAME, index = 2)
+ private final String stateName;
+
+ @JsonCreator
+ public StateMetadata(
+ @JsonProperty(FIELD_NAME_STATE_INDEX) int stateIndex,
+ @JsonProperty(FIELD_NAME_STATE_TTL) long stateTtl,
+ @JsonProperty(FIELD_NAME_STATE_NAME) String stateName) {
Review Comment:
please check the `stateName ` is not null, otherwise `tateName.equals(that.stateName);` in `equals` method will cause NPE
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #22492: [FLINK-31950][table-planner] Introduce StateMetadata and StateMetadataJson SerDe
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #22492:
URL: https://github.com/apache/flink/pull/22492#discussion_r1193015588
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/StateMetadataJsonSerializer.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.serde;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.exec.StateMetadata;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer;
+
+import java.io.IOException;
+
+import static org.apache.flink.table.planner.plan.nodes.exec.StateMetadata.FIELD_NAME_STATE_INDEX;
+import static org.apache.flink.table.planner.plan.nodes.exec.StateMetadata.FIELD_NAME_STATE_NAME;
+import static org.apache.flink.table.planner.plan.nodes.exec.StateMetadata.FIELD_NAME_STATE_TTL;
+
+/**
+ * JSON serializer for {@link StateMetadata}.
+ *
+ * @see StateMetadataJsonDeserializer for the reverse operation.
+ */
+@Internal
+public class StateMetadataJsonSerializer extends StdSerializer<StateMetadata> {
+
+ private static final long serialVersionUID = 1L;
+
+ StateMetadataJsonSerializer() {
+ super(StateMetadata.class);
+ }
+
+ @Override
+ public void serialize(
+ StateMetadata stateMetadata,
+ JsonGenerator jsonGenerator,
+ SerializerProvider serializerProvider)
+ throws IOException {
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeNumberField(FIELD_NAME_STATE_INDEX, stateMetadata.getStateIndex());
+ jsonGenerator.writeStringField(FIELD_NAME_STATE_TTL, stateMetadata.getStateTtl() + " ms");
Review Comment:
Using `Duration` is a good idea, but if we omit the StateMetadataJsonSerializer/StateMetadataJsonDeserializer, the output format will be like "PT166H40M". Is it acceptable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on a diff in pull request #22492: [FLINK-31950][table-planner] Introduce StateMetadata and StateMetadataJson SerDe
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on code in PR #22492:
URL: https://github.com/apache/flink/pull/22492#discussion_r1193015588
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/StateMetadataJsonSerializer.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.serde;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.exec.StateMetadata;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer;
+
+import java.io.IOException;
+
+import static org.apache.flink.table.planner.plan.nodes.exec.StateMetadata.FIELD_NAME_STATE_INDEX;
+import static org.apache.flink.table.planner.plan.nodes.exec.StateMetadata.FIELD_NAME_STATE_NAME;
+import static org.apache.flink.table.planner.plan.nodes.exec.StateMetadata.FIELD_NAME_STATE_TTL;
+
+/**
+ * JSON serializer for {@link StateMetadata}.
+ *
+ * @see StateMetadataJsonDeserializer for the reverse operation.
+ */
+@Internal
+public class StateMetadataJsonSerializer extends StdSerializer<StateMetadata> {
+
+ private static final long serialVersionUID = 1L;
+
+ StateMetadataJsonSerializer() {
+ super(StateMetadata.class);
+ }
+
+ @Override
+ public void serialize(
+ StateMetadata stateMetadata,
+ JsonGenerator jsonGenerator,
+ SerializerProvider serializerProvider)
+ throws IOException {
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeNumberField(FIELD_NAME_STATE_INDEX, stateMetadata.getStateIndex());
+ jsonGenerator.writeStringField(FIELD_NAME_STATE_TTL, stateMetadata.getStateTtl() + " ms");
Review Comment:
Using `Duration` is a good idea, but if we omit the StateMetadataJsonSerializer/StateMetadataJsonDeserializer, the output format will be like "PT166H40M". Is it acceptable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe closed pull request #22492: [FLINK-31950][table-planner] Introduce StateMetadata and StateMetadataJson SerDe
Posted by "godfreyhe (via GitHub)" <gi...@apache.org>.
godfreyhe closed pull request #22492: [FLINK-31950][table-planner] Introduce StateMetadata and StateMetadataJson SerDe
URL: https://github.com/apache/flink/pull/22492
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] LadyForest commented on pull request #22492: [FLINK-31950][table-planner] Introduce StateMetadata and support operator-level state TTL configuration via Compiled Plan
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest commented on PR #22492:
URL: https://github.com/apache/flink/pull/22492#issuecomment-1543402081
Since it's a huge change, let's separate this PR into several smaller PRs to ease the review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org