gaborgsomogyi commented on code in PR #26933:
URL: https://github.com/apache/flink/pull/26933#discussion_r2294271446
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/module/StateModule.java:
##########
@@ -20,70 +20,46 @@
import org.apache.flink.annotation.Experimental;
import org.apache.flink.state.table.SavepointMetadataTableFunction;
-import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.functions.BuiltInFunctionDefinition;
+import
org.apache.flink.table.functions.DynamicBuiltInFunctionDefinitionFactory;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.module.Module;
-import org.apache.flink.table.types.inference.TypeStrategies;
-import java.util.Collections;
+import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
+import java.util.ServiceLoader;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
-import static org.apache.flink.table.functions.FunctionKind.TABLE;
-
/** Module of state in Flink. */
@Experimental
public class StateModule implements Module {
public static final String IDENTIFIER = "state";
- public static final BuiltInFunctionDefinition SAVEPOINT_METADATA =
- BuiltInFunctionDefinition.newBuilder()
- .name("savepoint_metadata")
- .kind(TABLE)
-
.runtimeClass(SavepointMetadataTableFunction.class.getName())
- .outputTypeStrategy(
- TypeStrategies.explicit(
- DataTypes.ROW(
- DataTypes.FIELD(
- "checkpoint-id",
DataTypes.BIGINT().notNull()),
- DataTypes.FIELD("operator-name",
DataTypes.STRING()),
- DataTypes.FIELD("operator-uid",
DataTypes.STRING()),
- DataTypes.FIELD(
- "operator-uid-hash",
-
DataTypes.STRING().notNull()),
- DataTypes.FIELD(
- "operator-parallelism",
- DataTypes.INT().notNull()),
- DataTypes.FIELD(
- "operator-max-parallelism",
- DataTypes.INT().notNull()),
- DataTypes.FIELD(
-
"operator-subtask-state-count",
- DataTypes.INT().notNull()),
- DataTypes.FIELD(
-
"operator-coordinator-state-size-in-bytes",
-
DataTypes.BIGINT().notNull()),
- DataTypes.FIELD(
-
"operator-total-size-in-bytes",
-
DataTypes.BIGINT().notNull()))))
- .build();
-
public static final StateModule INSTANCE = new StateModule();
private final Map<String, BuiltInFunctionDefinition> normalizedFunctions;
private final Set<String> functionNamesWithInternal;
private final Set<String> functionNamesWithoutInternal;
private StateModule() {
- final List<BuiltInFunctionDefinition> definitions =
- Collections.singletonList(SAVEPOINT_METADATA);
+ final List<BuiltInFunctionDefinition> definitions = new ArrayList<>();
+
+ definitions.add(SavepointMetadataTableFunction.SAVEPOINT_METADATA);
+ ServiceLoader.load(DynamicBuiltInFunctionDefinitionFactory.class)
+ .iterator()
+ .forEachRemaining(
+ f -> {
+ if (f.factoryIdentifier().startsWith(IDENTIFIER)) {
Review Comment:
I think it makes sense to use `startsWith(IDENTIFIER + ".")` in order to
avoid accidental `stateful-x` pull-in so modified.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
