morrySnow commented on code in PR #65173:
URL: https://github.com/apache/doris/pull/65173#discussion_r3772624725
##########
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java:
##########
@@ -317,6 +323,10 @@ transactionId, this, getTimeZone(), getTimeout(),
task.init(loadId, attachment.getFileStatusByTable(aggKey),
attachment.getFileNumByTable(aggKey), getUserInfo());
+ ConnectContext context = ConnectContext.get();
+ if (context != null) {
+
recordLoadBackendSelectionSummary(context.getBackendSelectionProfile().getLoadSummary());
Review Comment:
The load summary is captured before the coordinator is actually selected.
`task.init()` (line 324) runs the planner, during which
`FileLoadScanNode.applyLoadBackendSelection` records the *first
provider-ordered file-scan candidate* into the profile; the real coordinator is
only chosen later in `LoadLoadingTask.executeOnce` (via
`BackendDistributedPlanWorkerManager` /
`NereidsCoordinator.recordLoadSinkCoordinator`), after `createTask` has already
read `getLoadSummary()` here. Since
`BackendSelectionProfile.recordLoadCoordinator` is first-writer-wins, the
actual coordinator record is silently dropped and `SHOW LOAD`/audit reports the
first preferred file-scan candidate as `coordinator_backend=...` — which may
not be the coordinator at all, and may not even be alive (no availability check
at the recording site). Consider recording the coordinator after the
coordinator is chosen (e.g. inside `executeOnce`), or not treating the
scan-side candidate as the coordinator.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/FileLoadScanNode.java:
##########
@@ -101,6 +107,19 @@ public void finalizeForNereids(TUniqueId loadId,
List<NereidsFileGroupInfo> file
}
}
+ static void applyLoadBackendSelection(ConnectContext context,
FederationBackendPolicy backendPolicy)
+ throws UserException {
+ List<Backend> orderedBackends =
BackendSelectionManager.orderLoadCandidates(
+ context, ImmutableList.copyOf(backendPolicy.getBackends()));
+ backendPolicy.replaceBackendOrder(orderedBackends);
+ BackendSelection.SelectionHint hint =
BackendSelectionManager.resolveLoadSelectionHint(context);
+ if (context != null
+ && BackendSelectionManager.hasLoadSelectionPreference(hint)
+ && !orderedBackends.isEmpty()) {
+ context.getBackendSelectionProfile().recordLoadCoordinator(hint,
orderedBackends.get(0));
Review Comment:
`orderedBackends.get(0)` is recorded as the "coordinator backend" without
any availability check, and this list is the file-scan backend candidate list —
not the sink/coordinator backends. If the top preferred candidate is
dead/load-unavailable at execution time, the profile (and the persisted
`loadBackendSelectionSummary` in SHOW LOAD, since `recordLoadCoordinator` is
first-wins) shows a coordinator_backend that never coordinated. Two issues to
consider: (1) filter by availability (e.g. `Backend::isLoadAvailable`) before
recording; (2) the summary label says "coordinator" while this is actually the
preferred scan backend — either record at the real coordinator site only, or
rename the field.
##########
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/MysqlLoadManager.java:
##########
@@ -667,24 +668,37 @@ private HttpPut generateRequestForMySqlLoadV2(
return httpPut;
}
- private String selectBackendForMySqlLoad(String database, String table)
throws LoadException {
+ private String selectBackendForMySqlLoad(ConnectContext context, String
database, String table)
+ throws LoadException {
Backend backend = null;
if (Config.isCloudMode()) {
String clusterName = "";
try {
- clusterName = ConnectContext.get().getCloudCluster();
+ clusterName = context.getCloudCluster();
} catch (Exception e) {
LOG.warn("failed to get cloud cluster: " + e.getMessage());
throw new LoadException("failed to get cloud cluster: " + e);
}
backend = StreamLoadHandler.selectBackend(clusterName);
+ if (backend == null) {
+ throw new
LoadException(SystemInfoService.NO_BACKEND_LOAD_AVAILABLE_MSG
+ + ", cluster: " + clusterName);
+ }
} else {
BeSelectionPolicy policy = new
BeSelectionPolicy.Builder().needLoadAvailable().build();
- List<Long> backendIds =
Env.getCurrentSystemInfo().selectBackendIdsByPolicy(policy, 1);
+ // The backend selection policy may reorder all eligible
candidates.
+ List<Long> backendIds =
Env.getCurrentSystemInfo().selectBackendIdsByPolicy(policy, -1);
Review Comment:
Behavior change for the default (no provider) path: previously
`selectBackendIdsByPolicy(policy, 1)` with `enableRoundRobin=false` returned a
random backend from ALL candidates (the `number == 1` early return happens
before the same-host dedup). With `-1`, `!allowOnSameHost` (default) first
dedups to one backend per host, so in deployments with multiple BEs on the same
host, one BE per host is silently excluded from the candidate set and the
selection distribution shifts to host-level randomization. This affects the
default no-op path too, so it is not gated behind the extension. If the intent
is to preserve "original candidate order and placement behavior" for the
default provider, the dedup should not apply here.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/LoadBalanceScanWorkerSelector.java:
##########
@@ -243,6 +254,12 @@ private Map<DistributedPlanWorker, UninstancedScanSource>
selectForBucket(
private WorkerScanRanges selectScanReplicaAndMinWorkloadWorker(
TScanRangeLocations tabletLocation, long tabletBytes, boolean
orderedScanRangeLocations, long catalogId) {
List<TScanRangeLocation> replicaLocations =
tabletLocation.getLocations();
+ if (orderedScanRangeLocations) {
+ replicaLocations = Lists.newArrayList(replicaLocations);
+ Collections.sort(replicaLocations);
+ }
+ replicaLocations = orderLoadReplicas(replicaLocations, catalogId);
Review Comment:
Design question: in REQUIRED mode, `orderLoadCandidates` filters the
candidates to the preferred partition (or throws if empty), and this is applied
to the *source-table replicas* of INSERT-SELECT loads as well, because
`useLoadBackendSelection` is set for any plan containing an `OlapTableSink`. So
`INSERT INTO t_preferred SELECT * FROM t_other` fails ("No candidate satisfies
required backend selection key") whenever the source table's backends are
outside the preferred partition — even though the requirement presumably only
concerns where the data is *written*. If REQUIRED is meant to constrain only
the sink/coordinator backends, source-replica reads should not be filtered by
`orderLoadCandidates`; if it is intentional, it should be documented so
providers partition source-table backends into the preferred set too.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java:
##########
@@ -232,7 +233,15 @@ private static void logAuditLogImpl(ConnectContext ctx,
String origStmt, Stateme
} catch (ComputeGroupException e) {
LOG.warn("Failed to get cloud cluster", e);
}
- String cluster = Config.isCloudMode() ? cloudCluster : "";
+ // Load statements resolve their own hint at the scheduling sites and
record it on the
+ // context; prefer it over the scan-side query decision so load audits
are accurate.
+ BackendSelection.SelectionHint selectionHint =
ctx.getLoadBackendSelectionDecisionForAudit();
+ if (selectionHint == null) {
+ selectionHint = ctx.getQueryBackendSelectionDecisionForAudit();
+ }
+ // In cloud mode, compute_group keeps its existing cloud compute group
meaning. In integrated
+ // mode, resource groups provide compute affinity, so reuse
compute_group for the preferred group.
+ String cluster = Config.isCloudMode() ? cloudCluster :
selectionHint.getPreferredKey();
Review Comment:
User-visible semantic change to the existing `compute_group` audit field: in
non-cloud mode it now carries the backend-selection preferred key once any
provider records a hint (for plain queries it's the *query* selection key).
Downstream users of the audit log who query `compute_group` will start seeing
backend-selection keys mixed in with real compute groups. The default provider
keeps "" so the community behavior is unchanged, but with an extension
installed this is a breaking semantic change to an existing audit column —
worth calling out in the release notes/docs.
##########
fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java:
##########
@@ -317,12 +334,16 @@ private long
selectBackendForCloudGroupCommitInternal(long tableId, String clust
throw new LoadException("No suitable backend for cloud cluster=" +
cluster + ", backends = " + backendsInfo);
}
- private long selectBackendForLocalGroupCommitInternal(long tableId) throws
LoadException {
+ private long selectBackendForLocalGroupCommitInternal(long tableId,
+ @Nullable BackendSelection.SelectionHint selectionHint) throws
LoadException {
if (LOG.isDebugEnabled()) {
LOG.debug("group commit select be info, tableToBeMap {},
tablePressureMap {}",
tableToBeMap.asMap().toString(),
tableToPressureMap.asMap().toString());
}
- Long cachedBackendId = getCachedBackend(null, tableId);
+ boolean hasLoadSelectionPreference =
BackendSelectionManager.hasLoadSelectionPreference(selectionHint);
+ String cacheKey = buildLocalGroupCommitCacheKey(tableId,
selectionHint, hasLoadSelectionPreference);
+ Long cachedBackendId =
BackendSelectionManager.isRequiredSelection(selectionHint)
Review Comment:
Two things here: (1) For REQUIRE mode the cache read is bypassed, but
`selectAvailableBackend` still writes `tableToBeMap.put(tableId#key#REQUIRE,
...)`, creating entries that are never read (dead weight in a 10000-entry
cache, though bounded). (2) More importantly, when required selection yields no
available preferred backend, `getRandomLocalBackend` returns null and the
method falls through to the generic `"No suitable backend ..."` LoadException
(line 374), so users can't distinguish a required-selection failure from a
normal "no backend" failure — while other sites surface a clear "No available
candidate satisfies required backend selection key '...'". Consider calling
`BackendSelectionManager.ensureRequiredSelectionSatisfied(selectionHint,
false)` before the final throw (only when the hint is required) to keep the
error consistent across load paths.
##########
fe/fe-core/src/main/java/org/apache/doris/clone/TabletSchedCtx.java:
##########
@@ -666,12 +682,32 @@ public void chooseSrcReplica(Map<Long, PathSlot>
backendsWorkingSlots, long exce
continue;
}
setSrc(srcReplica);
+ repairSourceSelectionResult = selectionEnabled
+ ?
BackendSelectionManager.classifyRepairSource(replicaBeId, destBackendId,
+ tablet.getReplicas(), candidates)
+ :
BackendSelectionProvider.RepairSourceSelectionResult.DISABLED;
return;
}
throw new SchedException(Status.SCHEDULE_FAILED, SubCode.WAITING_SLOT,
"waiting for source replica's slot");
}
+ static List<Replica> orderRepairSourceCandidates(List<Replica> candidates,
long destBackendId)
+ throws SchedException {
+ try {
+ return new
ArrayList<>(BackendSelectionManager.orderRepairSourceCandidates(
+ candidates, destBackendId));
+ } catch (UserException e) {
+ throw new SchedException(Status.UNRECOVERABLE, e.getMessage());
Review Comment:
A provider contract violation (e.g. dropped/duplicated replica in
`orderRepairSourceCandidates`) escalates to
`SchedException(Status.UNRECOVERABLE)`, which removes the tablet from the
repair scheduler permanently. With the community no-op provider this is
unreachable (identity always passes validation), but a downstream provider bug
would silently stop repairs for all affected tablets. Consider a retryable
status (e.g. `SCHEDULE_FAILED` / `WAITING_SLOT`) so a transient provider
mistake doesn't permanently abandon tablet repairs.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
