zhuzhurk commented on code in PR #27333:
URL: https://github.com/apache/flink/pull/27333#discussion_r2715753463
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/TasksBalancedRequestSlotMatchingStrategy.java:
##########
@@ -130,32 +131,35 @@ public int comparePriority(PhysicalSlotElement left,
PhysicalSlotElement right)
public Collection<RequestSlotMatch> matchRequestsAndSlots(
Collection<? extends PhysicalSlot> slots,
Collection<PendingRequest> pendingRequests,
- Map<ResourceID, LoadingWeight> taskExecutorsLoad) {
+ Map<ResourceID, TaskExecutionLoad> taskExecutionLoadMap) {
ResourceRequestPreMappings resourceRequestPreMappings =
ResourceRequestPreMappings.createFrom(pendingRequests, slots);
if (!resourceRequestPreMappings.isMatchingFulfilled()) {
return Collections.emptyList();
}
final Collection<RequestSlotMatch> resultingMatches = new
ArrayList<>();
- final List<PendingRequest> sortedRequests =
sortByLoadingDescend(pendingRequests);
+ final List<PendingRequest> sortedRequests = sortDesc(pendingRequests);
Review Comment:
Maybe `sortByTaskExecutionLoadDesc` to clarify the method's behavior?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtil.scala:
##########
@@ -191,9 +191,64 @@ object FlinkRelOptUtil {
* @return
* The digest of given rel tree.
*/
+ // --- FLINK CUSTOM START: CorrelationId normalization for subplan reuse
+ private def normalizeCorrelationIds(rel: RelNode): RelNode = {
Review Comment:
The change seems to be unrelated?
--
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]
