gaborgsomogyi commented on code in PR #29011:
URL: https://github.com/apache/flink/pull/29011#discussion_r3991033274
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java:
##########
@@ -214,11 +214,14 @@ public JobInfo getJobInfo() {
@Override
public TaskInfo getTaskInfo() {
+ // The read job's parallelism is unrelated to the operator's
maxParallelism and may exceed
+ // it (e.g. windowAll() has maxParallelism == 1); cap it to satisfy
TaskInfoImpl's
+ // maxParallelism >= numberOfParallelSubtasks invariant.
return new TaskInfoImpl(
ctx.getTaskInfo().getTaskName(),
maxParallelism,
indexOfSubtask,
- ctx.getTaskInfo().getNumberOfParallelSubtasks(),
+ Math.min(ctx.getTaskInfo().getNumberOfParallelSubtasks(),
maxParallelism),
ctx.getTaskInfo().getAttemptNumber());
Review Comment:
Is this an unrelated but important fix?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java:
##########
@@ -262,56 +335,42 @@ private static CatalogTable buildKeyedCatalogTable(
if (windowType == null) {
schemaBuilder.primaryKeyNamed("PK_state_key", "state_key");
}
+ // No formal PK on the window table: Flink's PK validation for
arbitrary (including
+ // zero-field) ROW columns like state_window is unreliable. Row
identity is still
+ // conceptually (state_key, state_window).
Schema schema = schemaBuilder.build();
Map<String, String> options = buildBaseConnectorOptions(statePath,
operatorIdentifier);
- options.put(
- SavepointConnectorOptions.STATE_READER_MODE.key(),
- (windowType == null
- ?
SavepointConnectorOptions.StateReaderMode.KEYED
- :
SavepointConnectorOptions.StateReaderMode.WINDOWED)
- .toString());
+ if (windowType != null) {
+ options.put(
+ SavepointConnectorOptions.STATE_READER_MODE.key(),
+
SavepointConnectorOptions.StateReaderMode.WINDOWED.toString());
+ }
Review Comment:
Why not building this already like this?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java:
##########
@@ -697,6 +726,20 @@ private static List<ResolvedTable>
candidateTablesForOperator(
}
}
+ KeyedStateSchemaInfo windowSchemaInfo =
+ StateTableUtils.getWindowKeyedStateSchema(metadata, opId);
+ if (!windowSchemaInfo.stateSchemas.isEmpty()) {
+ candidates.add(new ResolvedTable(opId, StateReaderMode.WINDOWED));
+ for (Map.Entry<String, KeyedStateSchemaInfo.StateEntryInfo> entry :
+ windowSchemaInfo.stateSchemas.entrySet()) {
+ StateType stateType = entry.getValue().stateType;
+ if (stateType == StateType.LIST || stateType == StateType.MAP)
{
+ candidates.add(
+ new ResolvedTable(opId,
StateReaderMode.WINDOWED_FLAT, entry.getKey()));
+ }
Review Comment:
Is it fine to fall through silently?
--
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]
