morningman commented on code in PR #67010:
URL: https://github.com/apache/doris/pull/67010#discussion_r3859929500
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -405,10 +409,37 @@ private void installTaskSnapshots(StatementContext
statementContext) {
}
private void setComputeGroup(ConnectContext ctx) {
- String taskComputeGroup = taskContext.getComputeGroup();
- if (Config.isCloudMode() && !Strings.isNullOrEmpty(taskComputeGroup)) {
- ctx.setCloudCluster(taskComputeGroup);
+ if (!Config.isCloudMode()) {
+ return;
+ }
+ // 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 declared = mtmv == null ? null :
mtmv.getComputeGroup().orElse(null);
+ String effective = Strings.isNullOrEmpty(declared) ?
taskContext.getComputeGroup() : declared;
+ if (!Strings.isNullOrEmpty(effective)) {
+ ctx.setCloudCluster(effective);
+ }
+ }
+
+ /**
+ * Re-checks the binding before the refresh runs: the groups can be
dropped and privileges
+ * revoked while the MV exists, and without this the refresh would fail
later with an unrelated
+ * message.
+ *
+ * <p>The identity used here is whatever the refresh actually runs as,
which today is the
+ * hardcoded {@code admin} (see {@link
MTMVPlanUtil#createBasicMvContext}). That makes the
+ * privilege half of the check always pass; the existence half is what has
teeth right now. Once
+ * an MV carries a real owner, passing that owner here is the only change
needed.
+ */
+ private void checkBindingBeforeTask(ConnectContext ctx) throws
UserException {
Review Comment:
Done in cb03a33 — renamed to `checkComputeGroupBeforeTask`, both here and on
the util.
I also narrowed it so the name is no longer a half-truth: the workload group
part is gone. It turned out to be redundant rather than just out of scope —
both callers resolve the workload group a little later through
`WorkloadGroupMgr#getWorkloadGroup(ConnectContext)` (routine load in
`KafkaTaskInfo#createRoutineLoadTask:210`, an MV refresh in
`NereidsCoordinator:521`), and that already does the same `USAGE` check against
the same owner plus the same existence check in the same compute group
namespace (`WorkloadGroupMgr:144`). It was also never exercised: every case in
`ComputeGroupBindingUtilTest` passed a null workload group.
The only behaviour difference is that an MV whose workload group was dropped
now fails during execution rather than just before the task starts — still a
`UserException` that fails the task, just with the coordinator's message. Tell
me if you'd rather keep the earlier failure and I'll put the check back under a
name of its own.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -405,10 +409,37 @@ private void installTaskSnapshots(StatementContext
statementContext) {
}
private void setComputeGroup(ConnectContext ctx) {
- String taskComputeGroup = taskContext.getComputeGroup();
- if (Config.isCloudMode() && !Strings.isNullOrEmpty(taskComputeGroup)) {
- ctx.setCloudCluster(taskComputeGroup);
+ if (!Config.isCloudMode()) {
+ return;
+ }
Review Comment:
Done in cb03a33 — `ConnectContext#setCloudCluster` now returns early in
non-cloud mode, and the guard here is gone.
Checked that nothing reads the session value outside cloud mode before
sinking it: `getCloudCluster(boolean)` throws `NOT_CLOUD_MODE` on its first
line, `CloudCoordinator` is cloud-only, and `StmtExecutor`'s read is already
inside an `isCloudMode` branch. Every existing FE UT that asserts on
`setCloudCluster` sets `Config.cloud_unique_id` first, so they are unaffected.
I did leave two call sites alone, because their guard is load-bearing rather
than redundant: `CreateRoutineLoadInfo` passes
`ConnectContext.get().getCloudCluster()` as the argument, which throws in
non-cloud mode, and `RoutineLoadJob#plan` has an else branch. The remaining
redundant ones (`FrontendServiceImpl`, `LoadAction`, `KafkaTaskInfo`,
`StreamingInsertTask`) are pre-existing; I kept them out of this PR to avoid
widening a backport, but happy to clean them up here if you prefer.
--
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]
