github-actions[bot] commented on code in PR #69749:
URL: https://github.com/apache/doris/pull/69749#discussion_r4399202162
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java:
##########
@@ -599,13 +625,30 @@ public void checkJobProperties() throws UserException {
RoutineLoadJob.DEFAULT_LOAD_TO_SINGLE_TABLET,
LOAD_TO_SINGLE_TABLET + " should be a boolean");
+ String inputComputeGroupStr = jobProperties.get(COMPUTE_GROUP);
+ if (!StringUtils.isEmpty(inputComputeGroupStr)) {
+ // The name is always adopted, because the workload group check
below resolves in its
+ // namespace, but it is only validated when a user is actually
declaring it. On the
+ // metadata load path the declared group may have been dropped
long ago, and failing
+ // there would cancel the job permanently instead of pausing it -
see isReplay.
+ if (!isReplay) {
+
ComputeGroupBindingUtil.validateDeclaredComputeGroup(ConnectContext.get(),
inputComputeGroupStr);
+ }
+ this.computeGroupName = inputComputeGroupStr;
+ }
+
String inputWorkloadGroupStr = jobProperties.get(WORKLOAD_GROUP);
if (!StringUtils.isEmpty(inputWorkloadGroupStr)) {
ConnectContext tmpCtx = new ConnectContext();
tmpCtx.setCurrentUserIdentity(ConnectContext.get().getCurrentUserIdentity());
tmpCtx.getSessionVariable().setWorkloadGroup(inputWorkloadGroupStr);
if (Config.isCloudMode()) {
- tmpCtx.setCloudCluster(ConnectContext.get().getCloudCluster());
+ // A workload group lives in the namespace of a compute group:
the same name under a
+ // different compute group is a different group. So this
existence check must be done
+ // against the compute group the job will actually run in, not
the session's one.
+
tmpCtx.setCloudCluster(StringUtils.isEmpty(this.computeGroupName)
+ ? ConnectContext.get().getCloudCluster()
+ : this.computeGroupName);
Review Comment:
[P1] Skip workload-group resource resolution during metadata replay
For a job created with both `compute_group='cg_a'` and
`workload_group='wg_a'`, `setReplay(true)` skips the direct compute-group check
but still reaches this lookup with `cg_a`. If that group has been dropped,
renamed or scaled to zero, `getWorkloadGroup(tmpCtx)` calls
`ComputeGroupMgr.getComputeGroupByName` and throws.
`RoutineLoadJob.gsonPostProcess()` catches it and sets the job to final
`CANCELLED`, so bringing the group back cannot recover it with RESUME. This
also reparses the original group after an ALTER to another group. Honor replay
mode for the downstream resource lookup too; the new replay test only supplies
`compute_group` and misses this branch.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -331,6 +332,10 @@ public void run() throws JobException {
}
mtmvSchemaChangeVersion = mtmv.getSchemaChangeVersion();
ConnectContext ctx = MTMVPlanUtil.createMTMVContext(mtmv,
MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK);
+ // The planning done on this context (base table resolution, partition
calculation) must see
+ // the same compute group the refresh will execute in, otherwise the
workload group would be
+ // looked up in a different namespace.
+ applyComputeGroup(ctx);
Review Comment:
[P2] Read compute and workload groups as one binding snapshot
`createMTMVContext()` already reads the MV's workload group, while this
later call independently reads its compute group. The refresh worker can read
`wg_a`, then a SQL thread atomically ALTER both properties from `(cg_a, wg_a)`
to `(cg_b, wg_b)`, then the worker reads `cg_b`. If `wg_a` exists only in A,
the ensuing workload-group lookup fails in B despite both stored configurations
being valid. The task's `MTMVJob` lock does not serialize property ALTER, which
only takes the MV lock. Both execution context paths have the same separated
reads. Snapshot the related properties under one MV read lock and use that
snapshot for context setup and checking.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java:
##########
@@ -438,6 +439,13 @@ protected void setOptional(CreateRoutineLoadInfo info)
throws UserException {
if (!StringUtils.isEmpty(info.getWorkloadGroupName())) {
jobProperties.put(WORKLOAD_GROUP, info.getWorkloadGroupName());
}
+
+ // Only write the key when the user declared it. An absent key means
"not declared" and
+ // keeps the existing implicit resolution, which is also what later
versions rely on to
+ // tell a pinned group apart from an inherited one.
+ if (!StringUtils.isEmpty(info.getComputeGroupName())) {
+ jobProperties.put(COMPUTE_GROUP, info.getComputeGroupName());
Review Comment:
[P2] Wake the declared group before the first Kafka metadata request
With cloud auto-start enabled, a session on live `cg_a` can declare an
existing but `SUSPENDED` `cg_b`. Validation accepts its name, and this property
immediately makes Kafka metadata requests use B, but none of that path calls
`waitForAutoStart(B)` as `USE @B` does. A CREATE with custom partitions
therefore fails during its initial Kafka metadata/offset lookup with no live
backends; otherwise the scheduler's partition lookup pauses the job and
repeatedly retries without ever requesting B's resume. Activate the declared
target before the metadata/backend lookup, without changing the caller's
session group or waiting under the job lock.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java:
##########
@@ -287,6 +289,16 @@ private void checkJobProperties() throws UserException {
analyzedJobProperties.put(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE,
modeStr.toUpperCase());
}
+ if (jobProperties.containsKey(CreateRoutineLoadInfo.COMPUTE_GROUP)) {
+ String computeGroup =
jobProperties.get(CreateRoutineLoadInfo.COMPUTE_GROUP);
+ if (!StringUtil.isEmpty(computeGroup)) {
+ // Unlike workload group, the compute group can be fully
validated right here, so do
+ // it now instead of letting a bad name pause the whole job on
the next task.
+ ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx,
computeGroup);
+ analyzedJobProperties.put(CreateRoutineLoadInfo.COMPUTE_GROUP,
computeGroup);
Review Comment:
[P2] Use the requested group when converting offsets during ALTER
A combined `ALTER ROUTINE LOAD ... PROPERTIES('compute_group'='live_b') FROM
KAFKA('kafka_partitions'='0', 'kafka_offsets'='OFFSET_END')` validates B here,
but `KafkaRoutineLoadJob.modifyProperties()` calls `convertOffset()` before
installing these properties. That conversion obtains the current job's old
group and sends its metadata RPC there. If the old group has disappeared or has
no live backends, ALTER fails before rebinding even though B is usable.
Timestamp offsets have the same issue. Pass the validated requested group into
offset conversion, retaining the old group only when this ALTER supplies no new
binding.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -1680,6 +1684,34 @@ private void setupComputeGroup(ConnectContext ctx) {
}
}
+ private void applyComputeGroup(ConnectContext ctx) {
+ // A compute group declared on the MV pins every refresh, automatic or
manual, to that group.
+ // Only when the MV declares nothing does a manual REFRESH keep
borrowing the session's group,
+ // which is the behaviour every existing MV keeps.
+ String effective =
mtmv.getComputeGroup().orElse(taskContext.getComputeGroup());
+ if (!Strings.isNullOrEmpty(effective)) {
+ // A no-op outside cloud mode.
+ ctx.setCloudCluster(effective);
Review Comment:
[P2] Keep the declared group authoritative after query hints run
An MV with `PROPERTIES('compute_group'='cg_a') AS SELECT /*+
SET_VAR(compute_group='cg_b') */ k1 FROM base` still refreshes in B. For the
partition path, the plan is `InsertOverwrite ->
LogicalSelectHint(SET_VAR(cg_b)) -> Project -> Scan`: setup records/checks A,
then `EliminateLogicalSelectHint` reapplies the hint during command analysis
and overwrites the session field consumed by `CloudCoordinator.prepare()`. In
the IVM path, parsing the query after setup also overwrites it. The task
reports A although execution uses B. Reject conflicting hints or preserve the
binding independently of hint-mutated session state, and validate/report the
group the coordinator actually consumes.
--
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]
