fix [INTERNAL_ERROR]sync filter size meet error, filter: RuntimeFilter:

924060929 · Oct 21, 2024 · c88d597 · c88d597
1 parent 9a9b2cb
commit c88d597
Showing 1 changed file with 5 additions and 11 deletions.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/RuntimeFiltersThriftBuilder.java b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/RuntimeFiltersThriftBuilder.java
@@ -98,22 +98,16 @@ public void setRuntimeFilterThriftParams(TRuntimeFilterParams runtimeFilterParam
                 for (RuntimeFilterTarget target : targets) {
                     targetParams.add(new TRuntimeFilterTargetParams(target.instanceId, target.address));
                 }
-                runtimeFilterParams.putToRidToTargetParam(
-                        rf.getFilterId().asInt(), targetParams
-                );
+                runtimeFilterParams.putToRidToTargetParam(rf.getFilterId().asInt(), targetParams);
             }
         }
         for (Map.Entry<RuntimeFilterId, Integer> entry : ridToBuilderNum.entrySet()) {
             boolean isBroadcastRuntimeFilter = broadcastRuntimeFilterIds.contains(entry.getKey().asInt());
             int builderNum = isBroadcastRuntimeFilter ? 1 : entry.getValue();
-            runtimeFilterParams.putToRuntimeFilterBuilderNum(
-                    entry.getKey().asInt(), builderNum
-            );
+            runtimeFilterParams.putToRuntimeFilterBuilderNum(entry.getKey().asInt(), builderNum);
         }
         for (RuntimeFilter rf : runtimeFilters) {
-            runtimeFilterParams.putToRidToRuntimeFilter(
-                    rf.getFilterId().asInt(), rf.toThrift()
-            );
+            runtimeFilterParams.putToRidToRuntimeFilter(rf.getFilterId().asInt(), rf.toThrift());
         }
     }
 
@@ -122,7 +116,7 @@ public static RuntimeFiltersThriftBuilder compute(
         PipelineDistributedPlan topMostPlan = distributedPlans.get(distributedPlans.size() - 1);
         AssignedJob mergeInstance = topMostPlan.getInstanceJobs().get(0);
         BackendWorker worker = (BackendWorker) mergeInstance.getAssignedWorker();
-        TNetworkAddress mergeAddress = new TNetworkAddress(worker.host(), worker.port());
+        TNetworkAddress mergeAddress = new TNetworkAddress(worker.host(), worker.brpcPort());
 
         List<RuntimeFilter> runtimeFilters = planner.getRuntimeFilters();
         Set<Integer> broadcastRuntimeFilterIds = runtimeFilters
@@ -144,7 +138,7 @@ public static RuntimeFiltersThriftBuilder compute(
                     Backend backend = backendWorker.getBackend();
                     targetFragments.add(new RuntimeFilterTarget(
                             instanceJob.instanceId(),
-                            new TNetworkAddress(backend.getHost(), backend.getBePort())
+                            new TNetworkAddress(backend.getHost(), backend.getBrpcPort())
                     ));
                 }
             }