This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 7ebf8c2751a branch-4.1: [feature](cloud) Support declaring
compute_group on routine load and async MV (#67010)
7ebf8c2751a is described below
commit 7ebf8c2751a634d4b3cadcebb13f82260490e9f7
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Sep 7 11:11:01 2026 +0800
branch-4.1: [feature](cloud) Support declaring compute_group on routine
load and async MV (#67010)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
Background jobs cannot be pinned to a specific compute group today. A
routine load
job silently snapshots whatever compute group the creating session
happened to be
on, and an async materialized view has no compute group property at all:
automatic
refreshes run wherever `admin` resolves to, while a manual `REFRESH`
borrows the
triggering session's group. So the same MV can refresh in two different
places
depending on who triggered it, and there is no way to isolate one job's
resources
from another's.
This PR adds a `compute_group` property to `CREATE`/`ALTER ROUTINE LOAD`
and to
async materialized views, so a job can be pinned explicitly.
This is a transitional binding ahead of the full
`(owner, compute_group, workload_group)` model. The property name and
its value
space are deliberately identical to that design, and the key is written
only when
the user actually declared it, so metadata written here is read back as
an explicit
pin later on with no conversion:
- the declaration lives in the already existing property maps
(`RoutineLoadJob.jobProperties`, `MTMV.mvProperties`), so no new
persisted field,
no journal type and no `FeMetaVersion` bump is needed;
- an absent key keeps the existing implicit resolution untouched, which
is also how
a pinned group is told apart from an inherited one;
- `DEFAULT` is rejected because it is reserved to mean "follow the
owner's default
group", and a group literally named `DEFAULT` would otherwise be
silently
reinterpreted after an upgrade;
- non-cloud is out of scope for now and rejected at `CREATE`/`ALTER`, so
non-cloud
metadata never carries the key.
**Routine load** resolves the effective cluster in `getCloudCluster()`,
which every
consumer already goes through: backend selection, the thrift task sent
to BE, the
workload group namespace, the plan context used by `OlapTableSink`, and
the
`ComputeGroup` column of `SHOW ROUTINE LOAD`. `ALTER` only updates the
property map
and is covered by the existing edit log, so replay needs no change.
**Async MVs** resolve it in `MTMVTask.setComputeGroup()`, the single
point where the
group is applied, so every trigger path is covered including the
scheduler one. A
declared group also wins over the triggering session for a manual
`REFRESH`, so the
same MV no longer refreshes in two different places depending on who
triggered it.
MVs that declare nothing keep borrowing the session's group, so existing
MVs are
unaffected.
The binding is also re-checked before every task, not only at create
time, because the
groups can be dropped and the owner's privileges revoked while a job
keeps running:
- **routine load** checks against the job's owner
(`RoutineLoadJob.userIdentity`, the
real creator), so revoking that user's `USAGE` on the pinned compute
group fails the
next task and pauses the job with the real reason, instead of surfacing
later as
"no available BE found";
- **async MVs** run the same check. Their refresh identity is still the
hardcoded
`admin`, so the privilege half always passes today; the existence half
is what has
teeth. Once an MV carries a real owner, passing that owner in is the
only change
needed.
Only the **explicitly declared** compute group is re-checked. A job that
declared none
is bound to whatever cluster its creating session happened to be on -
something the
user never chose - so putting that implicit binding under a new
privilege check would
start pausing jobs that predate this feature. Workload group behaviour
is unchanged:
routine load already re-checked it per task against the same owner.
Two correctness fixes fall out of this:
- `CREATE ROUTINE LOAD` validated the workload group against the
session's compute
group. Since a workload group lives in a compute group's namespace, that
check has
to run against the declared group, otherwise a valid combination is
rejected and an
invalid one only fails on the first task.
- `RoutineLoadJob.plan()` set the plan context from the raw snapshot
field rather
than the getter, which would have sent the write path to the creating
session's
group while the tasks ran in the declared one.
- `ComputeGroupMgr.getComputeGroupByName()` builds its "not found"
message from the
thread-local `ConnectContext`, which the routine load scheduler and the
MV task
runner do not have, so a dropped group would have surfaced as an NPE
there. The
runtime check therefore tests existence through `getCloudClusterNames()`
instead.
Usage:
```sql
CREATE ROUTINE LOAD db.job ON tbl
COLUMNS TERMINATED BY ","
PROPERTIES ("compute_group" = "cg_etl")
FROM KAFKA (...);
ALTER ROUTINE LOAD FOR db.job PROPERTIES ("compute_group" = "cg_etl_2");
CREATE MATERIALIZED VIEW mv
BUILD DEFERRED REFRESH AUTO ON MANUAL
DISTRIBUTED BY RANDOM BUCKETS 2
PROPERTIES ("replication_num" = "1", "compute_group" = "cg_batch")
AS SELECT k1, k2 FROM tbl;
ALTER MATERIALIZED VIEW mv SET ("compute_group" = "cg_batch_2");
```
### Release note
Support declaring `compute_group` on routine load jobs and async
materialized views
in cloud mode, so a background job can be pinned to a specific compute
group instead
of inheriting one implicitly.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- Behavior changed:
- [ ] No.
- [x] Yes, twice, both scoped to jobs that declare the new property:
1. A materialized view that declares `compute_group` now refreshes in
that
group even when a manual `REFRESH` is triggered from a session on a
different compute group. MVs that do not declare the property are
unaffected and keep borrowing the session's group.
2. A job pinned to a compute group now fails its next task if that group
is
dropped or if its owner's `USAGE` on it is revoked. Jobs that declare no
compute group are unaffected.
- Does this need documentation?
- [ ] No.
- [x] Yes. The new `compute_group` property on routine load and async MV
needs a
doc entry, including that it is cloud-mode only for now and that
`DEFAULT` is
reserved.
### Check List (For Reviewer who merge this PR)
- [x] Confirm the release note
- [x] Confirm test cases
- [x] Confirm document
- [x] Add branch pick label
---------
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../main/java/org/apache/doris/catalog/MTMV.java | 18 +
.../doris/cloud/load/CloudRoutineLoadManager.java | 7 +-
.../apache/doris/common/util/PropertyAnalyzer.java | 1 +
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 41 +-
.../doris/load/routineload/KafkaTaskInfo.java | 4 +
.../doris/load/routineload/RoutineLoadJob.java | 35 +-
.../doris/load/routineload/RoutineLoadManager.java | 5 +-
.../load/routineload/RoutineLoadTaskScheduler.java | 17 +-
.../org/apache/doris/mtmv/MTMVPropertyUtil.java | 17 +
.../plans/commands/AlterRoutineLoadCommand.java | 16 +-
.../plans/commands/info/CreateRoutineLoadInfo.java | 45 +-
.../java/org/apache/doris/qe/ConnectContext.java | 5 +
.../computegroup/ComputeGroupBindingUtil.java | 161 +++++++
.../routineload/RoutineLoadComputeGroupTest.java | 499 +++++++++++++++++++++
.../apache/doris/mtmv/MTMVComputeGroupTest.java | 181 ++++++++
.../computegroup/ComputeGroupBindingUtilTest.java | 238 ++++++++++
.../test_compute_group_binding_isolation.groovy | 133 ++++++
.../test_routine_load_compute_group.groovy | 286 ++++++++++++
.../suites/mtmv_p0/test_mtmv_compute_group.groovy | 142 ++++++
19 files changed, 1837 insertions(+), 14 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
index 82955035282..fe2cf90f74f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
@@ -320,6 +320,24 @@ public class MTMV extends OlapTable {
}
}
+ /**
+ * The compute group explicitly declared on this MV, empty when the user
did not declare one.
+ * An empty result keeps the existing implicit resolution (admin's group
for auto refresh,
+ * the session's group for a manual REFRESH).
+ */
+ public Optional<String> getComputeGroup() {
+ readMvLock();
+ try {
+ if
(mvProperties.containsKey(PropertyAnalyzer.PROPERTIES_COMPUTE_GROUP) &&
!StringUtils
+
.isEmpty(mvProperties.get(PropertyAnalyzer.PROPERTIES_COMPUTE_GROUP))) {
+ return
Optional.of(mvProperties.get(PropertyAnalyzer.PROPERTIES_COMPUTE_GROUP));
+ }
+ return Optional.empty();
+ } finally {
+ readMvUnlock();
+ }
+ }
+
public boolean isUseForRewrite() {
readMvLock();
try {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudRoutineLoadManager.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudRoutineLoadManager.java
index cee1a079424..e7257475441 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudRoutineLoadManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudRoutineLoadManager.java
@@ -40,7 +40,12 @@ public class CloudRoutineLoadManager extends
RoutineLoadManager {
@Override
public void addRoutineLoadJob(RoutineLoadJob routineLoadJob, String
dbName, String tableName)
throws UserException {
- if (!Strings.isNullOrEmpty(ConnectContext.get().getCloudCluster())) {
+ // When the job declares a compute group explicitly, that declaration
is the binding and the
+ // session's cluster must not overwrite it.
+ if (!Strings.isNullOrEmpty(routineLoadJob.getDeclaredComputeGroup())) {
+ LOG.info("routine load job {} pinned to declared compute group {}",
+ routineLoadJob.getName(),
routineLoadJob.getDeclaredComputeGroup());
+ } else if
(!Strings.isNullOrEmpty(ConnectContext.get().getCloudCluster())) {
routineLoadJob.setCloudCluster(ConnectContext.get().getCloudCluster());
} else {
throw new UserException("cloud cluster is empty, please specify
cloud cluster");
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java
b/fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java
index 631bdfd812d..56edd5a4492 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java
@@ -222,6 +222,7 @@ public class PropertyAnalyzer {
"async_mv.query_rewrite.consistency_relaxed_tables";
public static final String PROPERTIES_REFRESH_PARTITION_NUM =
"refresh_partition_num";
public static final String PROPERTIES_WORKLOAD_GROUP = "workload_group";
+ public static final String PROPERTIES_COMPUTE_GROUP = "compute_group";
public static final String PROPERTIES_PARTITION_SYNC_LIMIT =
"partition_sync_limit";
public static final String PROPERTIES_PARTITION_TIME_UNIT =
"partition_sync_time_unit";
public static final String PROPERTIES_PARTITION_DATE_FORMAT =
"partition_date_format";
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
index f3dbabf97c9..ef84d6071ca 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
@@ -66,6 +66,7 @@ import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.QeProcessorImpl;
import org.apache.doris.qe.QueryState.MysqlStateType;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.resource.computegroup.ComputeGroupBindingUtil;
import org.apache.doris.system.SystemInfoService;
import org.apache.doris.thrift.TCell;
import org.apache.doris.thrift.TRow;
@@ -338,6 +339,7 @@ public class MTMVTask extends AbstractTask {
try {
setComputeGroup(ctx);
recordComputeGroup(ctx);
+ checkComputeGroupBeforeTask(ctx);
installTaskSnapshots(statementContext);
TUniqueId queryId = generateQueryId();
lastQueryId = DebugUtil.printId(queryId);
@@ -376,6 +378,10 @@ public class MTMVTask extends AbstractTask {
private ConnectContext createTaskContext() {
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.
+ setComputeGroup(ctx);
ctx.setStatementContext(new StatementContext());
return ctx;
}
@@ -405,10 +411,39 @@ public class MTMVTask extends AbstractTask {
}
private void setComputeGroup(ConnectContext ctx) {
- String taskComputeGroup = taskContext.getComputeGroup();
- if (Config.isCloudMode() && !Strings.isNullOrEmpty(taskComputeGroup)) {
- ctx.setCloudCluster(taskComputeGroup);
+ // 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 = declared;
+ // A task read back from meta carries no taskContext, which the class
already tolerates
+ // elsewhere, so the session's group is only consulted when there is
one.
+ if (Strings.isNullOrEmpty(effective) && taskContext != null) {
+ effective = taskContext.getComputeGroup();
}
+ if (!Strings.isNullOrEmpty(effective)) {
+ ctx.setCloudCluster(effective);
+ }
+ }
+
+ /**
+ * Re-checks the declared compute group before the refresh runs: it can be
dropped and its
+ * 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 checkComputeGroupBeforeTask(ConnectContext ctx) throws
UserException {
+ if (mtmv == null) {
+ return;
+ }
+ // Only the explicitly declared compute group is re-checked; an MV
that declares none borrows
+ // the session's group, which the user never chose for it.
+
ComputeGroupBindingUtil.checkComputeGroupBeforeTask(ctx.getCurrentUserIdentity(),
+ mtmv.getComputeGroup().orElse(null));
}
private void recordComputeGroup(ConnectContext ctx) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
index 6b8c7636a00..78d4312812f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
@@ -82,6 +82,10 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
public TRoutineLoadTask createRoutineLoadTask() throws UserException {
KafkaRoutineLoadJob routineLoadJob = (KafkaRoutineLoadJob)
routineLoadManager.getJob(jobId);
+ // The declared compute group is re-checked before every task, but
that happens earlier, in
+ // RoutineLoadTaskScheduler#scheduleOneTask: it has to run before
backend allocation and
+ // before beginTxn, neither of which has happened by the time this
method is called.
+
// init tRoutineLoadTask and create plan fragment
TRoutineLoadTask tRoutineLoadTask = new TRoutineLoadTask();
TUniqueId queryId = new TUniqueId(id.getMostSignificantBits(),
id.getLeastSignificantBits());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
index 51c2ca42a4d..c0f98aa2960 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
@@ -129,6 +129,7 @@ public abstract class RoutineLoadJob
protected static final String STAR_STRING = "*";
public static final String WORKLOAD_GROUP = "workload_group";
+ public static final String COMPUTE_GROUP = "compute_group";
@Getter
@Setter
@@ -434,6 +435,13 @@ public abstract class RoutineLoadJob
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());
+ }
}
protected void setRoutineLoadDesc(RoutineLoadDesc routineLoadDesc) {
@@ -548,6 +556,11 @@ public abstract class RoutineLoadJob
return jobProperties.get(WORKLOAD_GROUP);
}
+ // The compute group explicitly declared on the job, or null when the user
did not declare one.
+ public String getDeclaredComputeGroup() {
+ return jobProperties.get(COMPUTE_GROUP);
+ }
+
public JobState getState() {
return state;
}
@@ -795,7 +808,12 @@ public abstract class RoutineLoadJob
}
public String getCloudCluster() {
- return cloudCluster;
+ // An explicitly declared compute group wins over the cluster
snapshotted from the session
+ // at create time. Keeping this in the getter (instead of writing both
places) means every
+ // caller follows the declaration, and ALTER only has to update
jobProperties, which is
+ // already covered by the existing edit log.
+ String declared = getDeclaredComputeGroup();
+ return StringUtils.isEmpty(declared) ? cloudCluster : declared;
}
public int getSizeOfRoutineLoadTaskInfoList() {
@@ -1056,13 +1074,16 @@ public abstract class RoutineLoadJob
table.readLock();
try {
if (Config.isCloudMode()) {
+ // Use the effective cluster, not the raw snapshot field: this
context is what
+ // OlapTableSink uses to pick the backends the load writes to.
+ String effectiveCluster = getCloudCluster();
if (ConnectContext.get() == null) {
ConnectContext ctx = new ConnectContext();
ctx.setThreadLocalInfo();
- ctx.setCloudCluster(cloudCluster);
+ ctx.setCloudCluster(effectiveCluster);
needCleanCtx = true;
} else {
- ConnectContext.get().setCloudCluster(cloudCluster);
+ ConnectContext.get().setCloudCluster(effectiveCluster);
}
ConnectContext.get().setCurrentUserIdentity(this.getUserIdentity());
} else {
@@ -1664,7 +1685,9 @@ public abstract class RoutineLoadJob
}
public String getClusterInfo() {
- return Strings.nullToEmpty(cloudCluster);
+ // SHOW ROUTINE LOAD must display the cluster the job actually runs
in, i.e. the declared
+ // compute group when there is one.
+ return Strings.nullToEmpty(getCloudCluster());
}
// check the correctness of commit info
@@ -2006,6 +2029,10 @@ public abstract class RoutineLoadJob
CreateRoutineLoadCommand command = (CreateRoutineLoadCommand)
nereidsParser.parseSingle(
origStmt.originStmt);
CreateRoutineLoadInfo createRoutineLoadInfo =
command.getCreateRoutineLoadInfo();
+ // This re-parse only rebuilds the RoutineLoadDesc; it must
not re-check resources
+ // that can legitimately have disappeared since the job was
created, because the
+ // catch below turns any failure into the final CANCELLED
state.
+ createRoutineLoadInfo.setReplay(true);
// If tableId is set, resolve the current table name by ID so
that
// table rename / SWAP TABLE won't cause replay to fail with
stale name in origStmt.
if (!isMultiTable && tableId != 0) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
index de52bb8d459..6ef436e0b37 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
@@ -505,8 +505,9 @@ public class RoutineLoadManager implements Writable {
if (availableBeIds.isEmpty()) {
RoutineLoadJob job = getJob(jobId);
if (job != null) {
- String msg = "no available BE found for job " + jobId + ",
cluster Name {}, " + job.getCloudCluster()
- + "please check the BE status and user's cluster or
tags";
+ String msg = "no available BE found for job " + jobId + ",
cluster name: "
+ + job.getCloudCluster()
+ + ", please check the BE status and user's cluster or
tags";
job.updateState(RoutineLoadJob.JobState.PAUSED,
new ErrorReason(InternalErrorCode.INTERNAL_ERR, msg),
false /* not replay */);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
index e1c4ce2b3dc..1f20a9864ff 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
@@ -30,6 +30,7 @@ import org.apache.doris.common.util.LogBuilder;
import org.apache.doris.common.util.LogKey;
import org.apache.doris.common.util.MasterDaemon;
import org.apache.doris.load.routineload.RoutineLoadJob.JobState;
+import org.apache.doris.resource.computegroup.ComputeGroupBindingUtil;
import org.apache.doris.system.Backend;
import org.apache.doris.thrift.BackendService;
import org.apache.doris.thrift.TNetworkAddress;
@@ -135,9 +136,23 @@ public class RoutineLoadTaskScheduler extends MasterDaemon
{
}
try {
- if
(routineLoadManager.getJob(routineLoadTaskInfo.getJobId()).isFinal()) {
+ RoutineLoadJob job =
routineLoadManager.getJob(routineLoadTaskInfo.getJobId());
+ if (job.isFinal()) {
return;
}
+
+ // The compute group the job declared can be dropped and the
owner's USAGE on it revoked
+ // while the job keeps running, so it is re-checked before every
task.
+ //
+ // This has to happen here rather than while the task is being
built. Backend allocation
+ // below resolves the backends through that same compute group, so
a missing group would
+ // otherwise pause the job with a generic "no available BE found"
before this check was
+ // ever reached, which is exactly the misleading message it exists
to replace. Running it
+ // first also keeps a failing check from leaving a transaction
behind, because beginTxn
+ // has not happened yet.
+
ComputeGroupBindingUtil.checkComputeGroupBeforeTask(job.getUserIdentity(),
+ job.getDeclaredComputeGroup());
+
// check if topic has more data to consume
if (!routineLoadTaskInfo.hasMoreDataToConsume()) {
needScheduleTasksQueue.addLast(routineLoadTaskInfo);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
index eeae843aae6..2d007d97815 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
@@ -18,10 +18,12 @@
package org.apache.doris.mtmv;
import org.apache.doris.catalog.Env;
+import org.apache.doris.common.UserException;
import org.apache.doris.common.util.PropertyAnalyzer;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.resource.computegroup.ComputeGroupBindingUtil;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
@@ -36,6 +38,7 @@ public class MTMVPropertyUtil {
PropertyAnalyzer.ASYNC_MV_QUERY_REWRITE_CONSISTENCY_RELAXED_TABLES,
PropertyAnalyzer.PROPERTIES_REFRESH_PARTITION_NUM,
PropertyAnalyzer.PROPERTIES_WORKLOAD_GROUP,
+ PropertyAnalyzer.PROPERTIES_COMPUTE_GROUP,
PropertyAnalyzer.PROPERTIES_PARTITION_SYNC_LIMIT,
PropertyAnalyzer.PROPERTIES_PARTITION_TIME_UNIT,
PropertyAnalyzer.PROPERTIES_PARTITION_DATE_FORMAT,
@@ -60,6 +63,9 @@ public class MTMVPropertyUtil {
case PropertyAnalyzer.PROPERTIES_WORKLOAD_GROUP:
analyzeWorkloadGroup(value);
break;
+ case PropertyAnalyzer.PROPERTIES_COMPUTE_GROUP:
+ analyzeComputeGroup(value);
+ break;
case PropertyAnalyzer.PROPERTIES_PARTITION_TIME_UNIT:
analyzePartitionTimeUnit(value);
break;
@@ -107,6 +113,17 @@ public class MTMVPropertyUtil {
}
}
+ private static void analyzeComputeGroup(String value) {
+ if (StringUtils.isEmpty(value)) {
+ return;
+ }
+ try {
+
ComputeGroupBindingUtil.validateDeclaredComputeGroup(ConnectContext.get(),
value);
+ } catch (UserException e) {
+ throw new AnalysisException(e.getMessage(), e);
+ }
+ }
+
private static void analyzeWorkloadGroup(String value) {
if (StringUtils.isEmpty(value)) {
return;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
index 367480c5d93..0ddf658ff85 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
@@ -40,6 +40,7 @@ import
org.apache.doris.nereids.trees.plans.commands.load.LoadProperty;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.resource.computegroup.ComputeGroupBindingUtil;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
@@ -74,6 +75,7 @@ public class AlterRoutineLoadCommand extends AlterCommand {
.add(CreateRoutineLoadInfo.STRICT_MODE)
.add(CreateRoutineLoadInfo.TIMEZONE)
.add(CreateRoutineLoadInfo.WORKLOAD_GROUP)
+ .add(CreateRoutineLoadInfo.COMPUTE_GROUP)
.add(JsonFileFormatProperties.PROP_JSON_PATHS)
.add(JsonFileFormatProperties.PROP_STRIP_OUTER_ARRAY)
.add(JsonFileFormatProperties.PROP_NUM_AS_STRING)
@@ -162,7 +164,7 @@ public class AlterRoutineLoadCommand extends AlterCommand {
labelNameInfo.validate(ctx);
FeNameFormat.checkCommonName(NAME_TYPE, labelNameInfo.getLabel());
// check routine load job properties include desired concurrent number
etc.
- checkJobProperties();
+ checkJobProperties(ctx);
// check load properties
RoutineLoadJob job = Env.getCurrentEnv().getRoutineLoadManager()
.getJob(getDbName(), getJobName());
@@ -177,7 +179,7 @@ public class AlterRoutineLoadCommand extends AlterCommand {
}
}
- private void checkJobProperties() throws UserException {
+ private void checkJobProperties(ConnectContext ctx) throws UserException {
Optional<String> optional = jobProperties.keySet().stream().filter(
entity ->
!CONFIGURABLE_JOB_PROPERTIES_SET.contains(entity)).findFirst();
if (optional.isPresent()) {
@@ -287,6 +289,16 @@ public class AlterRoutineLoadCommand extends AlterCommand {
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);
+ }
+ }
+
if (jobProperties.containsKey(CreateRoutineLoadInfo.WORKLOAD_GROUP)) {
String workloadGroup =
jobProperties.get(CreateRoutineLoadInfo.WORKLOAD_GROUP);
if (!StringUtil.isEmpty(workloadGroup)) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
index f67ac2fd965..e231ede470e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
@@ -55,6 +55,7 @@ import
org.apache.doris.nereids.trees.plans.commands.load.LoadSequenceClause;
import org.apache.doris.nereids.trees.plans.commands.load.LoadWhereClause;
import org.apache.doris.nereids.util.PlanUtils;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.resource.computegroup.ComputeGroupBindingUtil;
import org.apache.doris.resource.workloadgroup.WorkloadGroup;
import org.apache.doris.thrift.TPartialUpdateNewRowPolicy;
import org.apache.doris.thrift.TUniqueKeyUpdateMode;
@@ -93,6 +94,7 @@ public class CreateRoutineLoadInfo {
public static final String PARTIAL_UPDATE_NEW_KEY_POLICY =
"partial_update_new_key_behavior";
public static final String UNIQUE_KEY_UPDATE_MODE =
"unique_key_update_mode";
public static final String WORKLOAD_GROUP = "workload_group";
+ public static final String COMPUTE_GROUP = "compute_group";
public static final String ENDPOINT_REGEX =
"[-A-Za-z0-9+&@#/%?=~_|!:,.;]+[-A-Za-z0-9+&@#/%=~_|]";
public static final String SEND_BATCH_PARALLELISM =
"send_batch_parallelism";
public static final String LOAD_TO_SINGLE_TABLET = "load_to_single_tablet";
@@ -129,6 +131,7 @@ public class CreateRoutineLoadInfo {
.add(PARTIAL_UPDATE_NEW_KEY_POLICY)
.add(UNIQUE_KEY_UPDATE_MODE)
.add(WORKLOAD_GROUP)
+ .add(COMPUTE_GROUP)
.add(FileFormatProperties.PROP_FORMAT)
.add(JsonFileFormatProperties.PROP_JSON_PATHS)
.add(JsonFileFormatProperties.PROP_STRIP_OUTER_ARRAY)
@@ -168,6 +171,16 @@ public class CreateRoutineLoadInfo {
private String workloadGroupName;
+ private String computeGroupName;
+
+ // Set only by the metadata load path (RoutineLoadJob#gsonPostProcess),
which re-parses the
+ // stored statement to rebuild the RoutineLoadDesc rather than to admit a
new job. Resource
+ // existence must not be re-checked there: a compute group can be dropped,
renamed or scaled to
+ // zero backends while a job exists, and an exception during metadata load
is turned into
+ // JobState.CANCELLED, which is final and cannot be undone by RESUME. Such
a job has to be
+ // paused by the per task check instead, which is recoverable.
+ private boolean isReplay = false;
+
/**
* support partial columns load(Only Unique Key Columns)
*/
@@ -382,6 +395,19 @@ public class CreateRoutineLoadInfo {
return workloadGroupName;
}
+ public String getComputeGroupName() {
+ return computeGroupName;
+ }
+
+ /**
+ * Marks this info object as rebuilt from persisted metadata instead of
parsed from a user
+ * statement, so that {@link #validate(ConnectContext)} skips checks
against resources that may
+ * legitimately have disappeared since the job was created.
+ */
+ public void setReplay(boolean isReplay) {
+ this.isReplay = isReplay;
+ }
+
/**
* analyze create table info
*/
@@ -609,13 +635,30 @@ public class CreateRoutineLoadInfo {
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);
}
List<WorkloadGroup> wgList =
Env.getCurrentEnv().getWorkloadGroupMgr()
.getWorkloadGroup(tmpCtx);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
index d05f8189161..6cec50d4ee5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
@@ -1552,6 +1552,11 @@ public class ConnectContext {
}
public void setCloudCluster(String cluster) {
+ // A compute group only exists in cloud mode. Swallowing the call here
instead of making
+ // every caller wrap it in `if (Config.isCloudMode())` keeps that
check in one place.
+ if (!Config.isCloudMode()) {
+ return;
+ }
this.getSessionVariable().setCloudCluster(cluster);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroupBindingUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroupBindingUtil.java
new file mode 100644
index 00000000000..c9f8044a183
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroupBindingUtil.java
@@ -0,0 +1,161 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.resource.computegroup;
+
+import org.apache.doris.analysis.ResourceTypeEnum;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.InternalErrorCode;
+import org.apache.doris.common.UserException;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.ConnectContext;
+
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * Validation for the {@code compute_group} property that can be declared on
background jobs
+ * (routine load / async materialized view).
+ *
+ * <p>This is a transitional binding: it adds the ability to <b>declare</b> a
compute group, and
+ * re-checks that declaration before every task, but does not change how the
group is resolved for
+ * jobs that declare none. The property name and its value space are
intentionally identical to the
+ * final {@code (owner, compute_group, workload_group)} design, so that
metadata written by this
+ * version can be read as an explicit "pin" by later versions without any
conversion.
+ *
+ * <p>Two values are rejected on purpose:
+ * <ul>
+ * <li>Any value in non-cloud mode - non-cloud support is not part of this
transitional change,
+ * so no non-cloud metadata will ever carry this key.</li>
+ * <li>{@code DEFAULT} (case insensitive) - it is reserved by the final
design to mean
+ * "follow the owner's default group at runtime". Allowing a job to pin
a group literally
+ * named {@code DEFAULT} would silently change its behavior after
upgrading.</li>
+ * </ul>
+ */
+public class ComputeGroupBindingUtil {
+
+ /**
+ * Reserved value in the final binding design: "not pinned, follow the
owner's default group".
+ * Rejected here so that no job can pin a group literally named {@code
DEFAULT}.
+ */
+ public static final String RESERVED_DEFAULT = "DEFAULT";
+
+ public static final String ERR_NON_CLOUD =
+ "Property 'compute_group' is only supported in cloud mode for
now.";
+
+ private ComputeGroupBindingUtil() {
+ }
+
+ /**
+ * Validates a user declared compute group name.
+ *
+ * <p>An empty value means "not declared" and is treated as a no-op by the
caller, which must
+ * not write the key into the job's property map at all.
+ *
+ * @param ctx the context of the user executing CREATE / ALTER; privileges
are checked against
+ * this user, matching how {@code workload_group} is validated
today
+ * @param computeGroup the declared name
+ */
+ public static void validateDeclaredComputeGroup(ConnectContext ctx, String
computeGroup) throws UserException {
+ if (StringUtils.isEmpty(computeGroup)) {
+ return;
+ }
+
+ if (!Config.isCloudMode()) {
+ throw new UserException(ERR_NON_CLOUD);
+ }
+
+ if (RESERVED_DEFAULT.equalsIgnoreCase(computeGroup)) {
+ throw new UserException("'" + RESERVED_DEFAULT + "' is a reserved
value for property 'compute_group'"
+ + " and can not be used as a compute group name here.");
+ }
+
+ if (ctx == null) {
+ throw new UserException("Can not validate property 'compute_group'
without a connect context.");
+ }
+
+ // Same two checks, and the same order, as `USE @<compute group>`.
+ if
(!Env.getCurrentEnv().getAccessManager().checkCloudPriv(ctx.getCurrentUserIdentity(),
+ computeGroup, PrivPredicate.USAGE, ResourceTypeEnum.CLUSTER)) {
+ throw new UserException("USAGE denied to user '" +
ctx.getQualifiedUser()
+ + "' for compute group '" + computeGroup + "'");
+ }
+
+ if (!((CloudSystemInfoService)
Env.getCurrentSystemInfo()).getCloudClusterNames().contains(computeGroup)) {
+ throw new UserException("Compute group '" + computeGroup + "' not
found.");
+ }
+ }
+
+ /**
+ * Re-checks the compute group a job declared, before each of its tasks
runs.
+ *
+ * <p>Creation-time validation alone is not enough: the group can be
dropped and the owner's
+ * privileges can be revoked while the job keeps running, and without this
check the task would
+ * silently keep using a group its owner is no longer entitled to, or fail
much later with an
+ * unrelated message such as "no available BE found".
+ *
+ * <p>Everything is checked against {@code owner}, the identity the task
actually runs as, not
+ * against whoever created or last altered the job.
+ *
+ * <p>The workload group is deliberately out of scope here. Both callers
resolve it a little
+ * later through {@code WorkloadGroupMgr#getWorkloadGroup(ConnectContext)}
- routine load in
+ * {@code KafkaTaskInfo#createRoutineLoadTask}, an MV refresh in the
coordinator - and that
+ * already runs the same USAGE check against the same owner and the same
existence check in the
+ * same compute group namespace.
+ *
+ * @param owner the identity the task runs as
+ * @param computeGroup the compute group declared on the job; empty means
the job declared none
+ * and there is nothing to re-check
+ */
+ public static void checkComputeGroupBeforeTask(UserIdentity owner, String
computeGroup)
+ throws UserException {
+ if (owner == null) {
+ // Jobs created before the owner was persisted; nothing to check
them against.
+ return;
+ }
+
+ if (!Config.isCloudMode() || StringUtils.isEmpty(computeGroup)) {
+ return;
+ }
+
+ // Deliberately not ComputeGroupMgr.getComputeGroupByName() for the
existence check: when
+ // the group is missing that builds a hint message from the
thread-local ConnectContext,
+ // and the callers here are background threads that do not have one.
+ //
+ // A missing group is left on the default INTERNAL_ERR, which
RoutineLoadTaskScheduler
+ // reports as a retryable pause: the name can come back on its own,
because a compute group
+ // that is merely scaled to zero backends is removed from the cluster
map and re-added when
+ // it scales up again. Auto resume then picks the job up without an
operator.
+ if (!((CloudSystemInfoService)
Env.getCurrentSystemInfo()).getCloudClusterNames()
+ .contains(computeGroup)) {
+ throw new UserException("Compute group '" + computeGroup + "' not
found.");
+ }
+
+ // A revoked privilege is the opposite: somebody decided this owner
may no longer use this
+ // group, and nothing will undo that by itself. CANNOT_RESUME_ERR
keeps ScheduleRule from
+ // auto resuming the job, which would otherwise pause and resume it
every few minutes for
+ // as long as the grant is missing.
+ if (!Env.getCurrentEnv().getAccessManager().checkCloudPriv(owner,
computeGroup,
+ PrivPredicate.USAGE, ResourceTypeEnum.CLUSTER)) {
+ throw new UserException(InternalErrorCode.CANNOT_RESUME_ERR,
+ "USAGE denied to user '" + owner.getQualifiedUser()
+ + "' for compute group '" + computeGroup + "'");
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadComputeGroupTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadComputeGroupTest.java
new file mode 100644
index 00000000000..fd4fd9dc958
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadComputeGroupTest.java
@@ -0,0 +1,499 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.load.routineload;
+
+import org.apache.doris.analysis.ResourceTypeEnum;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.load.CloudRoutineLoadManager;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.InternalErrorCode;
+import org.apache.doris.common.LoadException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.load.loadv2.LoadTask;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import
org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.LabelNameInfo;
+import org.apache.doris.nereids.trees.plans.commands.load.LoadProperty;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentMap;
+import java.util.stream.Collectors;
+
+/**
+ * The transitional {@code compute_group} declaration on routine load jobs.
+ *
+ * <p>The declaration lives in {@code jobProperties} under the key {@code
compute_group}. It is only
+ * written when the user actually declared one, so an absent key keeps the old
behaviour of using the
+ * cluster snapshotted from the creating session, and later versions can tell
a pinned group apart
+ * from an inherited one just by looking at whether the key is present.
+ */
+public class RoutineLoadComputeGroupTest {
+
+ private static final String SESSION_SNAPSHOT_CG = "cg_from_session";
+ private static final String DECLARED_CG = "cg_declared";
+
+ private String originalDeployMode;
+ private String originalCloudUniqueId;
+ private SystemInfoService originalSystemInfo;
+ private AccessControllerManager originalAccessManager;
+
+ @Before
+ public void setUp() {
+ originalDeployMode = Config.deploy_mode;
+ originalCloudUniqueId = Config.cloud_unique_id;
+ originalSystemInfo = Env.getCurrentSystemInfo();
+ originalAccessManager = Env.getCurrentEnv().getAccessManager();
+ }
+
+ @After
+ public void tearDown() {
+ Config.deploy_mode = originalDeployMode;
+ Config.cloud_unique_id = originalCloudUniqueId;
+ Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo",
originalSystemInfo);
+ Deencapsulation.setField(Env.getCurrentEnv(), "accessManager",
originalAccessManager);
+ ConnectContext.remove();
+ }
+
+ /**
+ * Puts the FE into cloud mode with a known set of existing compute
groups, and a session that
+ * holds (or does not hold) USAGE on whatever it asks for.
+ */
+ private void enterCloudMode(boolean hasPriv, List<String>
existingClusters) {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+
+ AccessControllerManager accessManager =
Mockito.mock(AccessControllerManager.class);
+
Mockito.when(accessManager.checkCloudPriv(Mockito.any(UserIdentity.class),
Mockito.anyString(),
+ Mockito.any(PrivPredicate.class),
Mockito.any(ResourceTypeEnum.class))).thenReturn(hasPriv);
+ Deencapsulation.setField(Env.getCurrentEnv(), "accessManager",
accessManager);
+
+ Map<String, List<Backend>> clusterToBackends = Maps.newHashMap();
+ for (String cluster : existingClusters) {
+ clusterToBackends.put(cluster,
Lists.newArrayList(createBackend(10000L + clusterToBackends.size())));
+ }
+ Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo",
+ new ClusterAwareCloudSystemInfoService(clusterToBackends));
+
+ ConnectContext ctx = new ConnectContext();
+ ctx.setCurrentUserIdentity(UserIdentity.ADMIN);
+ ctx.setThreadLocalInfo();
+ }
+
+ private KafkaRoutineLoadJob newJob(String snapshotCluster, String
declaredComputeGroup) {
+ KafkaRoutineLoadJob job = new KafkaRoutineLoadJob();
+ // serialization walks this field, give it something non-null
+ job.setOrigStmt(new OriginStatement("CREATE ROUTINE LOAD test ON tbl",
0));
+ Deencapsulation.setField(job, "cloudCluster", snapshotCluster);
+ if (declaredComputeGroup != null) {
+ Map<String, String> jobProperties = Maps.newHashMap();
+ jobProperties.put(RoutineLoadJob.COMPUTE_GROUP,
declaredComputeGroup);
+ Deencapsulation.setField(job, "jobProperties", jobProperties);
+ }
+ return job;
+ }
+
+ @Test
+ public void testDeclaredComputeGroupWinsOverSessionSnapshot() {
+ KafkaRoutineLoadJob job = newJob(SESSION_SNAPSHOT_CG, DECLARED_CG);
+ Assert.assertEquals(DECLARED_CG, job.getDeclaredComputeGroup());
+ Assert.assertEquals(DECLARED_CG, job.getCloudCluster());
+ }
+
+ // No declaration must leave the existing behaviour completely untouched.
+ @Test
+ public void testFallsBackToSessionSnapshotWhenNotDeclared() {
+ KafkaRoutineLoadJob job = newJob(SESSION_SNAPSHOT_CG, null);
+ Assert.assertNull(job.getDeclaredComputeGroup());
+ Assert.assertEquals(SESSION_SNAPSHOT_CG, job.getCloudCluster());
+ }
+
+ // An empty declared value is treated as "not declared", same as the
workload group property.
+ @Test
+ public void testEmptyDeclarationFallsBackToSnapshot() {
+ KafkaRoutineLoadJob job = newJob(SESSION_SNAPSHOT_CG, "");
+ Assert.assertEquals(SESSION_SNAPSHOT_CG, job.getCloudCluster());
+ }
+
+ // SHOW ROUTINE LOAD must report the cluster the job actually runs in.
+ @Test
+ public void testClusterInfoShowsEffectiveComputeGroup() {
+ Assert.assertEquals(DECLARED_CG, newJob(SESSION_SNAPSHOT_CG,
DECLARED_CG).getClusterInfo());
+ Assert.assertEquals(SESSION_SNAPSHOT_CG, newJob(SESSION_SNAPSHOT_CG,
null).getClusterInfo());
+ Assert.assertEquals("", newJob(null, null).getClusterInfo());
+ }
+
+ // Metadata compatibility: the declaration is carried in the already
existing jobProperties map,
+ // so it survives a serialization round trip without any new persisted
field.
+ @Test
+ public void testMetadataRoundTripKeepsDeclaration() {
+ KafkaRoutineLoadJob job = newJob(SESSION_SNAPSHOT_CG, DECLARED_CG);
+ String json = GsonUtils.GSON.toJson(job);
+ Assert.assertTrue(json, json.contains(RoutineLoadJob.COMPUTE_GROUP));
+
+ KafkaRoutineLoadJob restored = GsonUtils.GSON.fromJson(json,
KafkaRoutineLoadJob.class);
+ Assert.assertEquals(DECLARED_CG, restored.getDeclaredComputeGroup());
+ Assert.assertEquals(DECLARED_CG, restored.getCloudCluster());
+ }
+
+ // Downgrade safety: metadata written without the key must still load and
behave as before.
+ @Test
+ public void testMetadataRoundTripWithoutDeclaration() {
+ KafkaRoutineLoadJob job = newJob(SESSION_SNAPSHOT_CG, null);
+ String json = GsonUtils.GSON.toJson(job);
+ Assert.assertFalse(json, json.contains(RoutineLoadJob.COMPUTE_GROUP));
+
+ KafkaRoutineLoadJob restored = GsonUtils.GSON.fromJson(json,
KafkaRoutineLoadJob.class);
+ Assert.assertNull(restored.getDeclaredComputeGroup());
+ Assert.assertEquals(SESSION_SNAPSHOT_CG, restored.getCloudCluster());
+ }
+
+ /**
+ * Isolation: two jobs pinned to different compute groups must be
scheduled onto disjoint sets of
+ * backends, so neither can consume the other's resources. The backend set
is resolved per job
+ * and on every task allocation, not once at create time.
+ */
+ @Test
+ public void testJobsWithDifferentComputeGroupsGetDisjointBackends() throws
LoadException {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+
+ Backend beInA1 = createBackend(10001L);
+ Backend beInA2 = createBackend(10002L);
+ Backend beInB = createBackend(10003L);
+ Map<String, List<Backend>> clusterToBackends = Maps.newHashMap();
+ clusterToBackends.put("cg_a", Lists.newArrayList(beInA1, beInA2));
+ clusterToBackends.put("cg_b", Lists.newArrayList(beInB));
+ Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo",
+ new ClusterAwareCloudSystemInfoService(clusterToBackends));
+
+ KafkaRoutineLoadJob jobA = newJob(SESSION_SNAPSHOT_CG, "cg_a");
+ KafkaRoutineLoadJob jobB = newJob(SESSION_SNAPSHOT_CG, "cg_b");
+ Map<Long, RoutineLoadJob> jobs = Maps.newHashMap();
+ jobs.put(1L, jobA);
+ jobs.put(2L, jobB);
+ TestCloudRoutineLoadManager manager = new
TestCloudRoutineLoadManager(jobs);
+
+ List<Long> backendsForA = manager.getAvailableBackendIdsForTest(1L);
+ List<Long> backendsForB = manager.getAvailableBackendIdsForTest(2L);
+
+ Assert.assertEquals(Lists.newArrayList(beInA1.getId(),
beInA2.getId()), backendsForA);
+ Assert.assertEquals(Lists.newArrayList(beInB.getId()), backendsForB);
+ Assert.assertTrue("jobs pinned to different compute groups must not
share backends",
+ backendsForA.stream().noneMatch(backendsForB::contains));
+ }
+
+ /**
+ * Isolation is driven by the declaration, not by the creating session:
both jobs were created in
+ * the same session (same snapshot cluster) yet land on different backends.
+ */
+ @Test
+ public void testDeclarationOverridesIdenticalCreatingSession() throws
LoadException {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+
+ Backend beInA = createBackend(10001L);
+ Backend beInSession = createBackend(10009L);
+ Map<String, List<Backend>> clusterToBackends = Maps.newHashMap();
+ clusterToBackends.put("cg_a", Lists.newArrayList(beInA));
+ clusterToBackends.put(SESSION_SNAPSHOT_CG,
Lists.newArrayList(beInSession));
+ Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo",
+ new ClusterAwareCloudSystemInfoService(clusterToBackends));
+
+ Map<Long, RoutineLoadJob> jobs = Maps.newHashMap();
+ jobs.put(1L, newJob(SESSION_SNAPSHOT_CG, "cg_a"));
+ jobs.put(2L, newJob(SESSION_SNAPSHOT_CG, null));
+ TestCloudRoutineLoadManager manager = new
TestCloudRoutineLoadManager(jobs);
+
+ Assert.assertEquals(Lists.newArrayList(beInA.getId()),
manager.getAvailableBackendIdsForTest(1L));
+ Assert.assertEquals(Lists.newArrayList(beInSession.getId()),
manager.getAvailableBackendIdsForTest(2L));
+ }
+
+ // Changing the declaration takes effect on the next task allocation, no
restart required.
+ @Test
+ public void testAlteringDeclarationMovesJobToAnotherComputeGroup() throws
LoadException {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+
+ Backend beInA = createBackend(10001L);
+ Backend beInB = createBackend(10002L);
+ Map<String, List<Backend>> clusterToBackends = Maps.newHashMap();
+ clusterToBackends.put("cg_a", Lists.newArrayList(beInA));
+ clusterToBackends.put("cg_b", Lists.newArrayList(beInB));
+ Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo",
+ new ClusterAwareCloudSystemInfoService(clusterToBackends));
+
+ KafkaRoutineLoadJob job = newJob(SESSION_SNAPSHOT_CG, "cg_a");
+ Map<Long, RoutineLoadJob> jobs = Maps.newHashMap();
+ jobs.put(1L, job);
+ TestCloudRoutineLoadManager manager = new
TestCloudRoutineLoadManager(jobs);
+ Assert.assertEquals(Lists.newArrayList(beInA.getId()),
manager.getAvailableBackendIdsForTest(1L));
+
+ // what ALTER ROUTINE LOAD ... PROPERTIES("compute_group" = "cg_b")
ends up doing
+ Map<String, String> jobProperties = Deencapsulation.getField(job,
"jobProperties");
+ jobProperties.put(RoutineLoadJob.COMPUTE_GROUP, "cg_b");
+
+ Assert.assertEquals(Lists.newArrayList(beInB.getId()),
manager.getAvailableBackendIdsForTest(1L));
+ }
+
+ /**
+ * Metadata load must not re-check the declared compute group.
+ *
+ * <p>{@link RoutineLoadJob#gsonPostProcess()} re-parses the stored CREATE
statement on every FE
+ * metadata load - a restart, and every checkpoint - only to rebuild the
RoutineLoadDesc, and it
+ * turns any failure into {@code JobState.CANCELLED}, which is final and
cannot be undone by
+ * RESUME. A compute group can be dropped, renamed, or scaled to zero
backends (which
+ * CloudSystemInfoService treats as dropped) while a job exists, so
validating it there would
+ * kill the job permanently instead of letting the per task check pause it
recoverably.
+ */
+ @Test
+ public void testMetadataLoadDoesNotValidateDeclaredComputeGroup() throws
UserException {
+ enterCloudMode(true, Lists.newArrayList("cg_live"));
+
+ CreateRoutineLoadInfo info = newCreateInfo("cg_dropped");
+ info.setReplay(true);
+
+ info.checkJobProperties();
+
+ // Only the resource check is skipped: the declaration itself is still
adopted, so the job
+ // keeps running in the group it was pinned to once the group comes
back.
+ Assert.assertEquals("cg_dropped", info.getComputeGroupName());
+ }
+
+ // Negative control for the test above: on the real CREATE path the same
value must still be
+ // rejected, otherwise the skip would have disabled validation everywhere.
+ @Test
+ public void testCreateStillRejectsComputeGroupThatDoesNotExist() {
+ enterCloudMode(true, Lists.newArrayList("cg_live"));
+
+ CreateRoutineLoadInfo info = newCreateInfo("cg_dropped");
+
+ UserException e = Assert.assertThrows(UserException.class,
info::checkJobProperties);
+ Assert.assertTrue(e.getMessage(),
+ e.getMessage().contains("Compute group 'cg_dropped' not
found."));
+ }
+
+ /**
+ * The per task re-check has to run before the task takes any resource.
+ *
+ * <p>Backend allocation resolves the backends through the very compute
group being checked, so
+ * a missing group makes it return an empty list and pause the job with a
generic
+ * "no available BE found for job ... please check the BE status and
user's cluster or tags".
+ * If the re-check ran after allocation it could never fire for a dropped
group, and the
+ * operator would be sent to look at BE health for what is really a
compute group that is gone.
+ * Running it first also means no transaction has been begun yet when it
fails.
+ */
+ @Test
+ public void testDroppedComputeGroupFailsTaskBeforeBackendAllocation() {
+ enterCloudMode(true, Lists.newArrayList("cg_live"));
+
+ RecordingKafkaRoutineLoadJob job = new RecordingKafkaRoutineLoadJob();
+ Deencapsulation.setField(job, "state",
RoutineLoadJob.JobState.RUNNING);
+ Deencapsulation.setField(job, "userIdentity", UserIdentity.ADMIN);
+ Map<String, String> jobProperties = Maps.newHashMap();
+ jobProperties.put(RoutineLoadJob.COMPUTE_GROUP, "cg_dropped");
+ Deencapsulation.setField(job, "jobProperties", jobProperties);
+
+ Map<Long, RoutineLoadJob> jobs = Maps.newHashMap();
+ jobs.put(1L, job);
+ RoutineLoadTaskScheduler scheduler =
+ new RoutineLoadTaskScheduler(new
TestCloudRoutineLoadManager(jobs));
+
+ ConcurrentMap<Integer, Long> partitionIdToOffset =
Maps.newConcurrentMap();
+ partitionIdToOffset.put(1, 100L);
+ KafkaTaskInfo taskInfo = new AlwaysReadyKafkaTaskInfo(new UUID(1, 1),
1L, partitionIdToOffset);
+
+ try {
+ Deencapsulation.invoke(scheduler, "scheduleOneTask", taskInfo);
+ // Without the check in front, scheduling reaches
allocateTaskToBe, which finds no
+ // backend for the dropped group, pauses the job with "no
available BE found" and
+ // returns normally - so reaching this line is itself the
regression.
+ Assert.fail("scheduling a task of a job pinned to a dropped
compute group must fail,"
+ + " last pause reason: " + job.pauseMsg);
+ } catch (Exception expected) {
+ // scheduleOneTask pauses the job and rethrows
+ }
+
+ Assert.assertEquals(RoutineLoadJob.JobState.PAUSED, job.newState);
+ Assert.assertNotNull("the job must be paused with a reason",
job.pauseMsg);
+ Assert.assertTrue(job.pauseMsg, job.pauseMsg.contains("Compute group
'cg_dropped' not found."));
+ Assert.assertFalse("the operator must not be sent to look at BE status
for a dropped group",
+ job.pauseMsg.contains("no available BE found"));
+
+ // A group can come back by itself - one that is only scaled to zero
backends leaves the
+ // cluster map and re-enters it when it scales up - so this pause has
to stay retryable.
+ Assert.assertNotEquals(InternalErrorCode.CANNOT_RESUME_ERR,
job.pauseCode);
+ Assert.assertTrue("a dropped group must still auto resume once it is
back",
+ ScheduleRule.isNeedAutoSchedule(job));
+ }
+
+ /**
+ * A revoked privilege is not transient, so the job must stop instead of
flapping.
+ *
+ * <p>Pausing with the default INTERNAL_ERR would let {@link ScheduleRule}
auto resume the job
+ * within at most MAX_BACK_OFF_TIME_SEC, whereupon the next task fails the
same check and pauses
+ * it again - for as long as the grant is missing, at two edit log entries
per cycle.
+ */
+ @Test
+ public void testRevokedUsagePausesTaskWithoutAutoResume() {
+ // the group exists, so the check gets past existence and fails on the
privilege
+ enterCloudMode(false, Lists.newArrayList("cg_pinned"));
+
+ RecordingKafkaRoutineLoadJob job = new RecordingKafkaRoutineLoadJob();
+ Deencapsulation.setField(job, "state",
RoutineLoadJob.JobState.RUNNING);
+ Deencapsulation.setField(job, "userIdentity", UserIdentity.ADMIN);
+ Map<String, String> jobProperties = Maps.newHashMap();
+ jobProperties.put(RoutineLoadJob.COMPUTE_GROUP, "cg_pinned");
+ Deencapsulation.setField(job, "jobProperties", jobProperties);
+
+ Map<Long, RoutineLoadJob> jobs = Maps.newHashMap();
+ jobs.put(1L, job);
+ RoutineLoadTaskScheduler scheduler =
+ new RoutineLoadTaskScheduler(new
TestCloudRoutineLoadManager(jobs));
+
+ ConcurrentMap<Integer, Long> partitionIdToOffset =
Maps.newConcurrentMap();
+ partitionIdToOffset.put(1, 100L);
+ KafkaTaskInfo taskInfo = new AlwaysReadyKafkaTaskInfo(new UUID(1, 1),
1L, partitionIdToOffset);
+
+ try {
+ Deencapsulation.invoke(scheduler, "scheduleOneTask", taskInfo);
+ Assert.fail("a task whose owner lost USAGE on the pinned group
must fail");
+ } catch (Exception expected) {
+ // scheduleOneTask pauses the job and rethrows
+ }
+
+ Assert.assertEquals(RoutineLoadJob.JobState.PAUSED, job.newState);
+ Assert.assertNotNull("the job must be paused with a reason",
job.pauseMsg);
+ Assert.assertTrue(job.pauseMsg, job.pauseMsg.contains("USAGE denied"));
+ Assert.assertEquals(InternalErrorCode.CANNOT_RESUME_ERR,
job.pauseCode);
+ Assert.assertFalse("a revoked privilege must not be auto resumed",
+ ScheduleRule.isNeedAutoSchedule(job));
+ }
+
+ private CreateRoutineLoadInfo newCreateInfo(String declaredComputeGroup) {
+ Map<String, String> jobProperties = Maps.newHashMap();
+ jobProperties.put(CreateRoutineLoadInfo.COMPUTE_GROUP,
declaredComputeGroup);
+ Map<String, String> dataSourceProperties = Maps.newHashMap();
+ dataSourceProperties.put("kafka_broker_list", "127.0.0.1:9092");
+ dataSourceProperties.put("kafka_topic", "test_topic");
+ Map<String, LoadProperty> loadPropertyMap = Maps.newHashMap();
+ return new CreateRoutineLoadInfo(new LabelNameInfo("test_db",
"test_job"), "test_tbl",
+ loadPropertyMap, jobProperties, "kafka", dataSourceProperties,
+ LoadTask.MergeType.APPEND, "");
+ }
+
+ private Backend createBackend(long id) {
+ Backend backend = new Backend(id, "127.0.0." + id, 9050);
+ backend.setAlive(true);
+ return backend;
+ }
+
+ /**
+ * KafkaTaskInfo asks the real RoutineLoadManager, and then Kafka, whether
there is more data to
+ * consume. Neither exists here, so answer yes and let scheduling proceed
to the part under
+ * test: without the compute group check in front, it must reach backend
allocation.
+ */
+ private static class AlwaysReadyKafkaTaskInfo extends KafkaTaskInfo {
+ private AlwaysReadyKafkaTaskInfo(UUID id, long jobId, Map<Integer,
Long> partitionIdToOffset) {
+ super(id, jobId, 20000, partitionIdToOffset, false, -1, false);
+ }
+
+ @Override
+ boolean hasMoreDataToConsume() {
+ return true;
+ }
+ }
+
+ /**
+ * Captures the state transition instead of writing an edit log, and
mirrors it onto the real
+ * fields so that ScheduleRule can be asked what it would do with the
resulting pause.
+ */
+ private static class RecordingKafkaRoutineLoadJob extends
KafkaRoutineLoadJob {
+ private JobState newState;
+ private String pauseMsg;
+ private InternalErrorCode pauseCode;
+
+ @Override
+ public void updateState(JobState jobState, ErrorReason reason, boolean
isReplay) {
+ this.newState = jobState;
+ this.pauseMsg = reason == null ? null : reason.getMsg();
+ this.pauseCode = reason == null ? null : reason.getCode();
+ this.state = jobState;
+ this.pauseReason = reason;
+ }
+ }
+
+ private static class ClusterAwareCloudSystemInfoService extends
CloudSystemInfoService {
+ private final Map<String, List<Backend>> clusterToBackends;
+
+ private ClusterAwareCloudSystemInfoService(Map<String, List<Backend>>
clusterToBackends) {
+ this.clusterToBackends = clusterToBackends;
+ }
+
+ @Override
+ public List<Backend> getBackendsByClusterName(final String
clusterName) {
+ return clusterToBackends.getOrDefault(clusterName,
Lists.newArrayList());
+ }
+
+ @Override
+ public List<String> getCloudClusterNames() {
+ return
clusterToBackends.keySet().stream().collect(Collectors.toList());
+ }
+ }
+
+ private static class TestCloudRoutineLoadManager extends
CloudRoutineLoadManager {
+ private final Map<Long, RoutineLoadJob> jobs;
+
+ private TestCloudRoutineLoadManager(Map<Long, RoutineLoadJob> jobs) {
+ this.jobs = jobs;
+ }
+
+ @Override
+ public RoutineLoadJob getJob(long jobId) {
+ return jobs.get(jobId);
+ }
+
+ // the real one looks the task up in idToRoutineLoadJob, which this
stub never fills
+ @Override
+ public boolean checkTaskInJob(RoutineLoadTaskInfo task) {
+ return jobs.containsKey(task.getJobId());
+ }
+
+ private List<Long> getAvailableBackendIdsForTest(long jobId) throws
LoadException {
+ return super.getAvailableBackendIds(jobId);
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVComputeGroupTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVComputeGroupTest.java
new file mode 100644
index 00000000000..bc9c60b7739
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVComputeGroupTest.java
@@ -0,0 +1,181 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.mtmv;
+
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.common.util.PropertyAnalyzer;
+import org.apache.doris.job.extensions.mtmv.MTMVTask;
+import org.apache.doris.job.extensions.mtmv.MTMVTask.MTMVTaskTriggerMode;
+import org.apache.doris.job.extensions.mtmv.MTMVTaskContext;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+
+/**
+ * The transitional {@code compute_group} declaration on async materialized
views.
+ *
+ * <p>A declared compute group pins every refresh - automatic and manual alike
- to that group.
+ * Only when nothing is declared does a manual REFRESH keep borrowing the
triggering session's group,
+ * which is the behaviour every existing MV keeps.
+ */
+public class MTMVComputeGroupTest {
+
+ private static final String DECLARED_CG = "cg_declared";
+ private static final String SESSION_CG = "cg_from_session";
+
+ private String originalDeployMode;
+ private String originalCloudUniqueId;
+
+ @Before
+ public void setUp() {
+ originalDeployMode = Config.deploy_mode;
+ originalCloudUniqueId = Config.cloud_unique_id;
+ }
+
+ @After
+ public void tearDown() {
+ Config.deploy_mode = originalDeployMode;
+ Config.cloud_unique_id = originalCloudUniqueId;
+ ConnectContext.remove();
+ }
+
+ private MTMV newMTMV(String declaredComputeGroup) {
+ MTMV mtmv = new MTMV();
+ Map<String, String> mvProperties = Maps.newHashMap();
+ if (declaredComputeGroup != null) {
+ mvProperties.put(PropertyAnalyzer.PROPERTIES_COMPUTE_GROUP,
declaredComputeGroup);
+ }
+ mtmv.setMvProperties(mvProperties);
+ return mtmv;
+ }
+
+ private MTMVTask newTask(MTMV mtmv, MTMVTaskTriggerMode triggerMode,
String taskContextComputeGroup) {
+ MTMVTaskContext taskContext = new MTMVTaskContext(
+ triggerMode, Lists.newArrayList(), false,
taskContextComputeGroup);
+ return new MTMVTask(mtmv, null, taskContext);
+ }
+
+ private String resolveComputeGroup(MTMVTask task) {
+ ConnectContext ctx = new ConnectContext();
+ Deencapsulation.invoke(task, "setComputeGroup", ctx);
+ return ctx.getSessionVariable().getCloudCluster();
+ }
+
+ @Test
+ public void testGetComputeGroupFromProperty() {
+ Assert.assertEquals(DECLARED_CG,
newMTMV(DECLARED_CG).getComputeGroup().orElse(null));
+ Assert.assertFalse(newMTMV(null).getComputeGroup().isPresent());
+ Assert.assertFalse(newMTMV("").getComputeGroup().isPresent());
+ }
+
+ @Test
+ public void testAlterComputeGroupProperty() {
+ MTMV mtmv = newMTMV(DECLARED_CG);
+ Map<String, String> changed = Maps.newHashMap();
+ changed.put(PropertyAnalyzer.PROPERTIES_COMPUTE_GROUP,
"cg_after_alter");
+ mtmv.alterMvProperties(changed);
+ Assert.assertEquals("cg_after_alter",
mtmv.getComputeGroup().orElse(null));
+ }
+
+ // Automatic refresh has no compute group on the task context; the
declaration is what pins it.
+ @Test
+ public void testAutoRefreshUsesDeclaredComputeGroup() {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+ MTMVTask task = newTask(newMTMV(DECLARED_CG),
MTMVTaskTriggerMode.SYSTEM, null);
+ Assert.assertEquals(DECLARED_CG, resolveComputeGroup(task));
+ }
+
+ // The one deliberate behaviour change: a declared compute group also wins
for a manual REFRESH,
+ // so the same MV never refreshes in two different places depending on who
triggered it.
+ @Test
+ public void testManualRefreshPrefersDeclaredOverSessionComputeGroup() {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+ MTMVTask task = newTask(newMTMV(DECLARED_CG),
MTMVTaskTriggerMode.MANUAL, SESSION_CG);
+ Assert.assertEquals(DECLARED_CG, resolveComputeGroup(task));
+ }
+
+ // Existing MVs declare nothing, so a manual REFRESH must keep borrowing
the session's group.
+ @Test
+ public void testManualRefreshFallsBackToSessionComputeGroup() {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+ MTMVTask task = newTask(newMTMV(null), MTMVTaskTriggerMode.MANUAL,
SESSION_CG);
+ Assert.assertEquals(SESSION_CG, resolveComputeGroup(task));
+ }
+
+ // Nothing declared and nothing on the task context leaves the context
untouched, which keeps the
+ // existing implicit resolution (admin's default compute group, or one
picked by policy).
+ @Test
+ public void testAutoRefreshWithoutDeclarationLeavesContextUntouched() {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+ MTMVTask task = newTask(newMTMV(null), MTMVTaskTriggerMode.SYSTEM,
null);
+ // untouched, i.e. left at the session variable's default so the
implicit resolution applies
+ Assert.assertTrue(Strings.isNullOrEmpty(resolveComputeGroup(task)));
+ }
+
+ // Non-cloud is out of scope; the declaration can not be created there,
and even if metadata
+ // somehow carried one it must not touch the context.
+ @Test
+ public void testNoopInNonCloudMode() {
+ Config.deploy_mode = "";
+ Config.cloud_unique_id = "";
+ MTMVTask task = newTask(newMTMV(DECLARED_CG),
MTMVTaskTriggerMode.MANUAL, SESSION_CG);
+ Assert.assertTrue(Strings.isNullOrEmpty(resolveComputeGroup(task)));
+ }
+
+ /**
+ * Isolation: two MVs declaring different compute groups resolve to
different clusters, which is
+ * what makes their refreshes pick disjoint backend sets downstream.
+ */
+ @Test
+ public void testMvsWithDifferentComputeGroupsResolveIndependently() {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+ MTMVTask taskA = newTask(newMTMV("cg_a"), MTMVTaskTriggerMode.SYSTEM,
null);
+ MTMVTask taskB = newTask(newMTMV("cg_b"), MTMVTaskTriggerMode.SYSTEM,
null);
+
+ String resolvedA = resolveComputeGroup(taskA);
+ String resolvedB = resolveComputeGroup(taskB);
+ Assert.assertEquals("cg_a", resolvedA);
+ Assert.assertEquals("cg_b", resolvedB);
+ Assert.assertNotEquals(resolvedA, resolvedB);
+ }
+
+ // A task carries no task context when it is read back from meta, so
resolving must fall back to
+ // the declaration rather than fail, and stay a no-op when there is no
declaration either.
+ @Test
+ public void testTaskWithoutTaskContextUsesDeclaredComputeGroup() {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+ Assert.assertEquals(DECLARED_CG, resolveComputeGroup(new
MTMVTask(newMTMV(DECLARED_CG), null, null)));
+ Assert.assertTrue(Strings.isNullOrEmpty(resolveComputeGroup(new
MTMVTask(newMTMV(null), null, null))));
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/resource/computegroup/ComputeGroupBindingUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/resource/computegroup/ComputeGroupBindingUtilTest.java
new file mode 100644
index 00000000000..d0c03b371d9
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/resource/computegroup/ComputeGroupBindingUtilTest.java
@@ -0,0 +1,238 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.resource.computegroup;
+
+import org.apache.doris.analysis.ResourceTypeEnum;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.InternalErrorCode;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.system.SystemInfoService;
+
+import com.google.common.collect.Lists;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.List;
+
+/**
+ * Validation of the transitional {@code compute_group} declaration.
+ */
+public class ComputeGroupBindingUtilTest {
+
+ private static final String CG_OK = "cg_ok";
+
+ private String originalDeployMode;
+ private String originalCloudUniqueId;
+ private SystemInfoService originalSystemInfo;
+ private SystemInfoService originalCgMgrSystemInfo;
+ private AccessControllerManager originalAccessManager;
+ private ConnectContext ctx;
+
+ @Before
+ public void setUp() {
+ originalDeployMode = Config.deploy_mode;
+ originalCloudUniqueId = Config.cloud_unique_id;
+ originalSystemInfo = Env.getCurrentSystemInfo();
+ originalCgMgrSystemInfo = Deencapsulation.getField(
+ Env.getCurrentEnv().getComputeGroupMgr(), "systemInfoService");
+ originalAccessManager = Env.getCurrentEnv().getAccessManager();
+ ctx = new ConnectContext();
+ ctx.setCurrentUserIdentity(UserIdentity.ADMIN);
+ }
+
+ @After
+ public void tearDown() {
+ Config.deploy_mode = originalDeployMode;
+ Config.cloud_unique_id = originalCloudUniqueId;
+ Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo",
originalSystemInfo);
+ Deencapsulation.setField(Env.getCurrentEnv().getComputeGroupMgr(),
"systemInfoService",
+ originalCgMgrSystemInfo);
+ Deencapsulation.setField(Env.getCurrentEnv(), "accessManager",
originalAccessManager);
+ ConnectContext.remove();
+ }
+
+ private void enterCloudMode(boolean hasPriv, List<String>
existingClusters) {
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+ AccessControllerManager accessManager =
Mockito.mock(AccessControllerManager.class);
+
Mockito.when(accessManager.checkCloudPriv(Mockito.any(UserIdentity.class),
Mockito.anyString(),
+ Mockito.any(PrivPredicate.class),
Mockito.any(ResourceTypeEnum.class))).thenReturn(hasPriv);
+ Deencapsulation.setField(Env.getCurrentEnv(), "accessManager",
accessManager);
+ TestCloudSystemInfoService svc = new
TestCloudSystemInfoService(existingClusters);
+ Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo", svc);
+ // ComputeGroupMgr captured its own reference at construction, so
replacing only Env's would
+ // leave getComputeGroupByName() casting the real non-cloud service.
+ Deencapsulation.setField(Env.getCurrentEnv().getComputeGroupMgr(),
"systemInfoService", svc);
+ }
+
+ // An empty declaration means "not declared" and must never fail, not even
in non-cloud mode,
+ // otherwise every existing job would break.
+ @Test
+ public void testEmptyDeclarationIsNoOp() throws UserException {
+ Config.deploy_mode = "";
+ Config.cloud_unique_id = "";
+ ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx, null);
+ ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx, "");
+ }
+
+ // Non-cloud is out of scope for this transitional change: declaring a
compute group must be
+ // rejected so that no non-cloud metadata ever carries the key.
+ @Test
+ public void testRejectedInNonCloudMode() {
+ Config.deploy_mode = "";
+ Config.cloud_unique_id = "";
+ UserException e = Assert.assertThrows(UserException.class,
+ () ->
ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx, CG_OK));
+ Assert.assertTrue(e.getMessage(), e.getMessage().contains("only
supported in cloud mode"));
+ }
+
+ // DEFAULT is reserved by the final binding design ("follow the owner's
default group").
+ // Pinning a group literally named DEFAULT would silently change behaviour
after upgrading.
+ @Test
+ public void testRejectReservedDefaultValue() {
+ enterCloudMode(true, Lists.newArrayList("DEFAULT", "default", CG_OK));
+ for (String reserved : new String[] {"DEFAULT", "default", "Default"})
{
+ UserException e = Assert.assertThrows(UserException.class,
+ () ->
ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx, reserved));
+ Assert.assertTrue(e.getMessage(),
e.getMessage().contains("reserved value"));
+ }
+ }
+
+ @Test
+ public void testRejectWhenNoUsagePrivilege() {
+ enterCloudMode(false, Lists.newArrayList(CG_OK));
+ UserException e = Assert.assertThrows(UserException.class,
+ () ->
ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx, CG_OK));
+ Assert.assertTrue(e.getMessage(), e.getMessage().contains("USAGE
denied"));
+ }
+
+ @Test
+ public void testRejectWhenComputeGroupDoesNotExist() {
+ enterCloudMode(true, Lists.newArrayList("some_other_cg"));
+ UserException e = Assert.assertThrows(UserException.class,
+ () ->
ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx, CG_OK));
+ Assert.assertTrue(e.getMessage(), e.getMessage().contains("not
found"));
+ }
+
+ @Test
+ public void testAcceptValidDeclaration() throws UserException {
+ enterCloudMode(true, Lists.newArrayList(CG_OK, "cg_other"));
+ ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx, CG_OK);
+ }
+
+ // The privilege check must run before the existence check, so that
probing for the existence of
+ // a compute group the user has no access to is not possible.
+ @Test
+ public void testPrivilegeIsCheckedBeforeExistence() {
+ enterCloudMode(false, Lists.newArrayList("some_other_cg"));
+ UserException e = Assert.assertThrows(UserException.class,
+ () ->
ComputeGroupBindingUtil.validateDeclaredComputeGroup(ctx, CG_OK));
+ Assert.assertTrue(e.getMessage(), e.getMessage().contains("USAGE
denied"));
+ Assert.assertFalse(e.getMessage(), e.getMessage().contains("not
found"));
+ }
+
+ // ---- checkComputeGroupBeforeTask: re-checked before every task, against
the job's owner ----
+
+ // Jobs created before the owner was persisted have nothing to check
against.
+ @Test
+ public void testRuntimeCheckSkippedWhenOwnerUnknown() throws UserException
{
+ enterCloudMode(false, Lists.newArrayList());
+ ComputeGroupBindingUtil.checkComputeGroupBeforeTask(null, "cg_gone");
+ }
+
+ // Nothing is bound, so there is nothing to re-check.
+ @Test
+ public void testRuntimeCheckSkippedWhenNothingBound() throws UserException
{
+ enterCloudMode(false, Lists.newArrayList());
+
ComputeGroupBindingUtil.checkComputeGroupBeforeTask(UserIdentity.ADMIN, null);
+ }
+
+ // The compute group was dropped while the job kept running.
+ //
+ // This runs without a thread-local ConnectContext on purpose: the callers
are background
+ // threads (the routine load scheduler, the MV task runner) that have
none, so the check must
+ // not reach any code path that needs one.
+ @Test
+ public void testRuntimeCheckFailsWhenComputeGroupDropped() {
+ enterCloudMode(true, Lists.newArrayList());
+ ConnectContext.remove();
+ UserException e = Assert.assertThrows(UserException.class,
+ () ->
ComputeGroupBindingUtil.checkComputeGroupBeforeTask(UserIdentity.ADMIN,
"cg_gone"));
+ Assert.assertTrue(e.getMessage(), e.getMessage().contains("not
found"));
+ // A missing group can come back on its own - one that is only scaled
to zero backends
+ // leaves the cluster map and re-enters it when it scales up - so the
pause this produces
+ // has to stay retryable for RoutineLoadTaskScheduler / ScheduleRule.
+ Assert.assertNotEquals(InternalErrorCode.CANNOT_RESUME_ERR,
e.getErrorCode());
+ }
+
+ // The owner's USAGE on the compute group was revoked while the job kept
running. This is the
+ // case creation-time validation alone cannot catch.
+ @Test
+ public void testRuntimeCheckFailsWhenComputeGroupUsageRevoked() {
+ enterCloudMode(false, Lists.newArrayList(CG_OK));
+ UserException e = Assert.assertThrows(UserException.class,
+ () ->
ComputeGroupBindingUtil.checkComputeGroupBeforeTask(UserIdentity.ADMIN, CG_OK));
+ Assert.assertTrue(e.getMessage(), e.getMessage().contains("USAGE
denied"));
+ // Unlike a missing group, nothing undoes a REVOKE by itself, so the
pause must not be auto
+ // resumed - otherwise the job flaps every few minutes for as long as
the grant is missing.
+ Assert.assertEquals(InternalErrorCode.CANNOT_RESUME_ERR,
e.getErrorCode());
+ }
+
+ // Non-cloud has no named compute group, so the check is skipped entirely.
+ @Test
+ public void testRuntimeCheckSkipsComputeGroupInNonCloudMode() throws
UserException {
+ Config.deploy_mode = "";
+ Config.cloud_unique_id = "";
+
ComputeGroupBindingUtil.checkComputeGroupBeforeTask(UserIdentity.ADMIN,
"cg_gone");
+ }
+
+ private static class TestCloudSystemInfoService extends
CloudSystemInfoService {
+ private final List<String> clusterNames;
+
+ private TestCloudSystemInfoService(List<String> clusterNames) {
+ this.clusterNames = clusterNames;
+ }
+
+ @Override
+ public List<String> getCloudClusterNames() {
+ return clusterNames;
+ }
+
+ // getComputeGroupByName() resolves through these two, so a known name
has to look
+ // resolvable here or the existence check fires before anything else
can be exercised.
+ @Override
+ public String getPhysicalCluster(String clusterName) {
+ return clusterName;
+ }
+
+ @Override
+ public String getCloudClusterIdByName(String clusterName) {
+ return clusterNames.contains(clusterName) ? clusterName + "_id" :
"";
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/test_compute_group_binding_isolation.groovy
b/regression-test/suites/cloud_p0/multi_cluster/test_compute_group_binding_isolation.groovy
new file mode 100644
index 00000000000..e2323b52cf4
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/test_compute_group_binding_isolation.groovy
@@ -0,0 +1,133 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.apache.doris.regression.suite.ClusterOptions
+
+// Isolation across compute groups, end to end.
+//
+// Two async materialized views declaring different compute groups must
refresh in their own group
+// and nowhere else, no matter which group the triggering session is on. The
MV refresh path needs
+// no external data source, which makes it the cheapest way to prove the
binding end to end.
+suite('test_compute_group_binding_isolation', 'multi_cluster,docker') {
+ def options = new ClusterOptions()
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ def secondGroup = "cg_isolation_b"
+
+ // Resolve the session's own compute group before the second group
exists. A session with no
+ // explicit group picks the first group with an alive backend, and
that list is sorted by
+ // name, so once cg_isolation_b is up it would win over the default
group and the two names
+ // below would collide.
+ def groups = sql_return_maparray """show clusters"""
+ logger.info("clusters: ${groups}")
+ def firstGroup = groups.stream()
+ .filter(cg -> cg.is_current ==
"TRUE").findFirst().orElse(null)?.cluster
+ assertNotNull(firstGroup)
+ assertNotEquals(firstGroup, secondGroup)
+
+ cluster.addBackend(1, secondGroup)
+ // Keep this session on its original group now that a second one
exists.
+ sql """use @${firstGroup}"""
+
+ def tableName = "test_cg_isolation_tbl"
+ def mvOnFirst = "test_cg_isolation_mv_a"
+ def mvOnSecond = "test_cg_isolation_mv_b"
+
+ sql """DROP TABLE IF EXISTS ${tableName} FORCE"""
+ sql """
+ CREATE TABLE ${tableName} (
+ `k1` INT NULL,
+ `k2` INT NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`k1`)
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 2
+ PROPERTIES ('replication_num' = '1');
+ """
+ sql """INSERT INTO ${tableName} VALUES (1, 1), (2, 2), (3, 3);"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvOnFirst}
+ BUILD DEFERRED REFRESH AUTO ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1', 'compute_group' =
'${firstGroup}')
+ AS SELECT k1, k2 FROM ${tableName};
+ """
+ sql """
+ CREATE MATERIALIZED VIEW ${mvOnSecond}
+ BUILD DEFERRED REFRESH AUTO ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1', 'compute_group' =
'${secondGroup}')
+ AS SELECT k1, k2 FROM ${tableName};
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ${mvOnFirst} AUTO;"""
+ sql """REFRESH MATERIALIZED VIEW ${mvOnSecond} AUTO;"""
+ waitingMTMVTaskFinishedByMvName(mvOnFirst)
+ waitingMTMVTaskFinishedByMvName(mvOnSecond)
+
+ def computeGroupOfLastTask = { mvName ->
+ def tasks = sql_return_maparray """
+ select * from tasks("type"="mv") where MvName = '${mvName}'
order by CreateTime desc
+ """
+ assertTrue(tasks.size() > 0)
+ logger.info("tasks of ${mvName}: ${tasks}")
+ return tasks.get(0).ComputeGroup
+ }
+
+ // Each MV refreshed in its own compute group, and the two are
different.
+ assertEquals(firstGroup, computeGroupOfLastTask(mvOnFirst))
+ assertEquals(secondGroup, computeGroupOfLastTask(mvOnSecond))
+
+ // Both produced correct data, i.e. pinning to a non-session group did
not break the refresh.
+ def rowsA = sql "SELECT COUNT(*) FROM ${mvOnFirst}"
+ def rowsB = sql "SELECT COUNT(*) FROM ${mvOnSecond}"
+ assertEquals(3, rowsA.get(0).get(0))
+ assertEquals(3, rowsB.get(0).get(0))
+
+ // The declaration wins over the triggering session: refreshing from
inside the second group
+ // must still send the first MV to the first group.
+ // COMPLETE, not AUTO: the base table has not changed since the first
refresh, so AUTO would
+ // resolve to NOT_REFRESH and never report a compute group to assert
on.
+ sql """use @${secondGroup}"""
+ sql """REFRESH MATERIALIZED VIEW ${mvOnFirst} COMPLETE;"""
+ waitingMTMVTaskFinishedByMvName(mvOnFirst)
+ assertEquals(firstGroup, computeGroupOfLastTask(mvOnFirst))
+
+ // And an MV that declares nothing keeps the old behaviour of
borrowing the session's group.
+ def mvUndeclared = "test_cg_isolation_mv_none"
+ sql """
+ CREATE MATERIALIZED VIEW ${mvUndeclared}
+ BUILD DEFERRED REFRESH AUTO ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k1, k2 FROM ${tableName};
+ """
+ sql """REFRESH MATERIALIZED VIEW ${mvUndeclared} AUTO;"""
+ waitingMTMVTaskFinishedByMvName(mvUndeclared)
+ assertEquals(secondGroup, computeGroupOfLastTask(mvUndeclared))
+
+ sql """use @${firstGroup}"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvOnFirst};"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvOnSecond};"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvUndeclared};"""
+ sql """DROP TABLE IF EXISTS ${tableName} FORCE"""
+ }
+}
diff --git
a/regression-test/suites/load_p0/routine_load/test_routine_load_compute_group.groovy
b/regression-test/suites/load_p0/routine_load/test_routine_load_compute_group.groovy
new file mode 100644
index 00000000000..4f3ba4f0032
--- /dev/null
+++
b/regression-test/suites/load_p0/routine_load/test_routine_load_compute_group.groovy
@@ -0,0 +1,286 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.clients.producer.ProducerConfig
+
+// Declaring `compute_group` on a routine load job.
+//
+// The declaration is stored in the job's property map under the key
`compute_group`, which is the
+// same key and value space the final (owner, compute_group, workload_group)
design uses, so the
+// metadata is read back as an explicit pin by later versions without
conversion.
+//
+// The property-level checks run before the Kafka data source is touched, so
the negative cases do
+// not need a broker; only the "job really runs there" cases do.
+suite("test_routine_load_compute_group", "p0") {
+ String tableName = "test_routine_load_compute_group_tbl"
+ String jobName = "test_routine_load_compute_group_job"
+
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `k1` INT NULL,
+ `k2` STRING NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`k1`)
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+
+ // Returns the statement instead of running it: inside a `test { }` block
the `sql` method must
+ // be the action's own, so the SQL has to be handed to it there rather
than executed here.
+ def createJobSql = { String computeGroup ->
+ return """
+ CREATE ROUTINE LOAD ${jobName} ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES ("compute_group" = "${computeGroup}")
+ FROM KAFKA ("kafka_broker_list" = "127.0.0.1:19092", "kafka_topic"
= "unused_topic");
+ """.toString()
+ }
+
+ if (!isCloudMode()) {
+ // Non-cloud is out of scope for this transitional change, so the key
can never appear in
+ // non-cloud metadata and upgrading such a cluster has nothing to
convert. The cloud-mode
+ // check runs first, so even the reserved DEFAULT is refused with this
message.
+ for (String value : ["any_group", "DEFAULT", "default"]) {
+ test {
+ sql createJobSql(value)
+ exception "only supported in cloud mode"
+ }
+ }
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ return
+ }
+
+ // ---------------- cloud mode ----------------
+
+ // DEFAULT is reserved by the final design ("follow the owner's default
group"). A job pinned to
+ // a group literally named DEFAULT would be silently reinterpreted after
an upgrade.
+ for (String reserved : ["DEFAULT", "default", "Default"]) {
+ test {
+ sql createJobSql(reserved)
+ exception "reserved value"
+ }
+ }
+
+ test {
+ sql createJobSql("cg_that_does_not_exist")
+ exception "not found"
+ }
+
+ def currentComputeGroup = sql_return_maparray("show clusters")
+ .stream().filter(cg -> cg.is_current ==
"TRUE").findFirst().orElse(null)
+ assertNotNull(currentComputeGroup)
+ def cgName = currentComputeGroup.cluster
+ logger.info("current compute group: ${cgName}")
+
+ String enabled = context.config.otherConfigs.get("enableKafkaTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("kafka test not enabled, skipping the running-job part")
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ return
+ }
+
+ String kafka_port = context.config.otherConfigs.get("kafka_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ def kafka_broker = "${externalEnvIp}:${kafka_port}"
+ def topic = "test_routine_load_compute_group"
+
+ def props = new Properties()
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"${kafka_broker}".toString())
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
+ "org.apache.kafka.common.serialization.StringSerializer")
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
+ "org.apache.kafka.common.serialization.StringSerializer")
+ props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000")
+ def producer = new KafkaProducer<>(props)
+ for (int i = 0; i < 5; i++) {
+ producer.send(new ProducerRecord<>(topic, "${i},row${i}".toString()))
+ }
+ producer.close()
+
+ try {
+ sql """
+ CREATE ROUTINE LOAD ${jobName} ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES (
+ "max_batch_interval" = "5",
+ "max_batch_rows" = "300000",
+ "max_batch_size" = "209715200",
+ "compute_group" = "${cgName}"
+ )
+ FROM KAFKA (
+ "kafka_broker_list" = "${kafka_broker}",
+ "kafka_topic" = "${topic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+
+ // SHOW must report the compute group the job actually runs in, and
the declaration must be
+ // visible in the job properties so operators can inventory pinned
jobs before an upgrade.
+ def show = sql_return_maparray("SHOW ROUTINE LOAD FOR
${jobName}").get(0)
+ logger.info("show routine load: ${show}")
+ assertEquals(cgName, show.ComputeGroup)
+ assertTrue(show.JobProperties.contains("compute_group"))
+ assertTrue(show.JobProperties.contains(cgName))
+
+ // The job runs, i.e. pinning did not break the load path.
+ def count = 0
+ while (count < 60) {
+ def state = sql_return_maparray("SHOW ROUTINE LOAD FOR
${jobName}").get(0).State
+ def rows = sql "SELECT COUNT(*) FROM ${tableName}"
+ logger.info("state=${state}, rows=${rows}")
+ assertNotEquals("PAUSED", state)
+ if (rows.get(0).get(0) >= 5) {
+ break
+ }
+ sleep(2000)
+ count++
+ }
+ assertEquals(5, sql("SELECT COUNT(*) FROM ${tableName}").get(0).get(0))
+
+ // ALTER validates the new value immediately instead of letting a bad
name pause the job on
+ // the next task. ALTER ROUTINE LOAD only accepts PAUSED jobs.
+ sql "PAUSE ROUTINE LOAD FOR ${jobName}"
+ test {
+ sql """ALTER ROUTINE LOAD FOR ${jobName}
PROPERTIES("compute_group" = "DEFAULT");"""
+ exception "reserved value"
+ }
+ test {
+ sql """ALTER ROUTINE LOAD FOR ${jobName}
PROPERTIES("compute_group" = "cg_nope");"""
+ exception "not found"
+ }
+
+ sql """ALTER ROUTINE LOAD FOR ${jobName} PROPERTIES("compute_group" =
"${cgName}");"""
+ def showAfterAlter = sql_return_maparray("SHOW ROUTINE LOAD FOR
${jobName}").get(0)
+ assertEquals(cgName, showAfterAlter.ComputeGroup)
+
+ sql "RESUME ROUTINE LOAD FOR ${jobName}"
+ } finally {
+ try {
+ sql "STOP ROUTINE LOAD FOR ${jobName}"
+ } catch (Exception e) {
+ logger.info("stop routine load failed: ${e.getMessage()}")
+ }
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ }
+
+ // The binding is re-checked before every task, not only at CREATE time:
revoking the job
+ // owner's USAGE on the declared compute group must pause the job with
that reason.
+ //
+ // The check is deliberately placed before the task takes any resource -
before backend
+ // allocation and before the load transaction is begun - so that a group
which has been dropped
+ // outright cannot first surface as "no available BE found". That ordering
is what
+ //
RoutineLoadComputeGroupTest#testDroppedComputeGroupFailsTaskBeforeBackendAllocation
pins
+ // down; here we cover the privilege half end to end, which needs a real
running job.
+ String revokeUser = "test_rl_cg_revoke_user"
+ String revokePwd = "Test_12345"
+ String revokeJob = "test_rl_cg_revoke_job"
+ String revokeTable = "test_rl_cg_revoke_tbl"
+
+ sql """DROP TABLE IF EXISTS ${revokeTable}"""
+ sql """
+ CREATE TABLE ${revokeTable} (
+ `k1` INT NULL,
+ `k2` STRING NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`k1`)
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+ sql """DROP USER IF EXISTS ${revokeUser}"""
+ sql """CREATE USER '${revokeUser}' IDENTIFIED BY '${revokePwd}'"""
+ sql """GRANT LOAD_PRIV, SELECT_PRIV ON *.*.* TO ${revokeUser}"""
+ sql """GRANT USAGE_PRIV ON COMPUTE GROUP `${cgName}` TO ${revokeUser}"""
+
+ try {
+ // Created by the user, so RoutineLoadJob#userIdentity - the identity
the per task check
+ // runs against - is that user rather than the admin running this
suite.
+ connect(revokeUser, "${revokePwd}", context.config.jdbcUrl) {
+ sql """use ${context.dbName}"""
+ sql """
+ CREATE ROUTINE LOAD ${revokeJob} ON ${revokeTable}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES (
+ "max_batch_interval" = "5",
+ "max_batch_rows" = "300000",
+ "max_batch_size" = "209715200",
+ "compute_group" = "${cgName}"
+ )
+ FROM KAFKA (
+ "kafka_broker_list" = "${kafka_broker}",
+ "kafka_topic" = "${topic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+ }
+
+ def runningBeforeRevoke = false
+ for (int i = 0; i < 60; i++) {
+ def state = sql_return_maparray("SHOW ROUTINE LOAD FOR
${revokeJob}").get(0).State
+ if (state == "RUNNING") {
+ runningBeforeRevoke = true
+ break
+ }
+ sleep(1000)
+ }
+ assertTrue(runningBeforeRevoke, "the job must be running before its
privilege is revoked")
+
+ sql """REVOKE USAGE_PRIV ON COMPUTE GROUP `${cgName}` FROM
${revokeUser}"""
+
+ // The check runs ahead of the "is there more data to consume" probe,
so the job is refused
+ // on the next scheduling round whether or not the topic has new
records.
+ def pausedForComputeGroup = false
+ def sawBeAvailabilityReason = false
+ def lastSeen = ""
+ for (int i = 0; i < 90; i++) {
+ def row = sql_return_maparray("SHOW ROUTINE LOAD FOR
${revokeJob}").get(0)
+ def reason = row.ReasonOfStateChanged == null ? "" :
row.ReasonOfStateChanged
+ lastSeen = "state=${row.State}, reason=${reason}"
+ if (reason.contains("no available BE found")) {
+ sawBeAvailabilityReason = true
+ }
+ if (reason.contains("USAGE denied") && reason.contains(cgName)) {
+ pausedForComputeGroup = true
+ break
+ }
+ sleep(1000)
+ }
+ assertTrue(pausedForComputeGroup,
+ "revoking USAGE must fail the next task with the real reason,
last seen: ${lastSeen}".toString())
+ assertFalse(sawBeAvailabilityReason,
+ "a compute group problem must never be reported as a BE
availability problem")
+
+ // Nothing undoes a REVOKE by itself, so the pause carries
CANNOT_RESUME_ERR and the job
+ // must stay down instead of being auto resumed into the same failure
every few minutes.
+ for (int i = 0; i < 15; i++) {
+ def state = sql_return_maparray("SHOW ROUTINE LOAD FOR
${revokeJob}").get(0).State
+ assertEquals("PAUSED", state,
+ "a job paused by a revoked privilege must not be auto
resumed")
+ sleep(1000)
+ }
+ } finally {
+ try {
+ sql "STOP ROUTINE LOAD FOR ${revokeJob}"
+ } catch (Exception e) {
+ logger.info("stop routine load failed: ${e.getMessage()}")
+ }
+ sql """DROP TABLE IF EXISTS ${revokeTable}"""
+ sql """DROP USER IF EXISTS ${revokeUser}"""
+ }
+}
diff --git a/regression-test/suites/mtmv_p0/test_mtmv_compute_group.groovy
b/regression-test/suites/mtmv_p0/test_mtmv_compute_group.groovy
new file mode 100644
index 00000000000..c671892a125
--- /dev/null
+++ b/regression-test/suites/mtmv_p0/test_mtmv_compute_group.groovy
@@ -0,0 +1,142 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Declaring `compute_group` on an async materialized view.
+//
+// The property is a transitional binding whose name and value space match the
final
+// (owner, compute_group, workload_group) design, so metadata written here is
read as an explicit
+// pin by later versions. Two values are rejected on purpose: anything in
non-cloud mode, and the
+// reserved word DEFAULT. The non-cloud check runs first, so in non-cloud mode
every value - DEFAULT
+// included - is refused with the same "cloud mode" message.
+suite("test_mtmv_compute_group") {
+ String suiteName = "test_mtmv_compute_group"
+ String tableName = "${suiteName}_table"
+ String mvName = "${suiteName}_mv"
+
+ sql """drop materialized view if exists ${mvName};"""
+ sql """drop table if exists `${tableName}`"""
+
+ sql """
+ CREATE TABLE `${tableName}` (
+ `k1` INT NULL,
+ `k2` INT NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`k1`)
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 2
+ PROPERTIES ('replication_num' = '1');
+ """
+ sql """insert into ${tableName} values(1, 1), (2, 2);"""
+
+ // Returns the statement instead of running it: inside a `test { }` block
the `sql` method must
+ // be the action's own, so the SQL has to be handed to it there rather
than executed here.
+ def createMvSql = { String properties ->
+ return """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH AUTO ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 2
+ PROPERTIES (${properties})
+ AS SELECT k1, k2 FROM ${tableName};
+ """.toString()
+ }
+
+ if (!isCloudMode()) {
+ // Non-cloud is out of scope for this transitional change: declaring
the property must be
+ // rejected whatever the value, so that non-cloud metadata never
carries the key and
+ // upgrading such a cluster has nothing to convert.
+ for (String value : ["any_group", "DEFAULT", "default"]) {
+ test {
+ sql createMvSql("'replication_num' = '1', 'compute_group' =
'${value}'")
+ exception "only supported in cloud mode"
+ }
+ }
+
+ // An MV without the property must keep working exactly as before.
+ sql createMvSql("'replication_num' = '1'")
+ sql """REFRESH MATERIALIZED VIEW ${mvName} AUTO;"""
+ waitingMTMVTaskFinishedByMvName(mvName)
+ assertEquals(2, sql("SELECT COUNT(*) FROM ${mvName}").get(0).get(0))
+
+ // ALTER must reject it too, not only CREATE.
+ test {
+ sql """ALTER MATERIALIZED VIEW ${mvName} SET ('compute_group' =
'any_group');"""
+ exception "only supported in cloud mode"
+ }
+
+ sql """drop materialized view if exists ${mvName};"""
+ sql """drop table if exists `${tableName}`"""
+ return
+ }
+
+ // ---------------- cloud mode ----------------
+
+ // DEFAULT is reserved by the final design ("follow the owner's default
group"). Pinning a group
+ // literally named DEFAULT would be silently reinterpreted after an
upgrade.
+ for (String reserved : ["DEFAULT", "default", "Default"]) {
+ test {
+ sql createMvSql("'replication_num' = '1', 'compute_group' =
'${reserved}'")
+ exception "reserved value"
+ }
+ }
+
+ test {
+ sql createMvSql("'replication_num' = '1', 'compute_group' =
'cg_that_does_not_exist'")
+ exception "not found"
+ }
+
+ def currentComputeGroup = sql_return_maparray("show clusters")
+ .stream().filter(cg -> cg.is_current ==
"TRUE").findFirst().orElse(null)
+ assertNotNull(currentComputeGroup)
+ def cgName = currentComputeGroup.cluster
+ logger.info("current compute group: ${cgName}")
+
+ sql createMvSql("'replication_num' = '1', 'compute_group' = '${cgName}'")
+
+ // The declaration is persisted on the MV and visible.
+ def showCreate = sql """show create materialized view ${mvName};"""
+ assertTrue(showCreate.toString().contains("compute_group"))
+ assertTrue(showCreate.toString().contains(cgName))
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} AUTO;"""
+ waitingMTMVTaskFinishedByMvName(mvName)
+ assertEquals(2, sql("SELECT COUNT(*) FROM ${mvName}").get(0).get(0))
+
+ // The refresh really ran in the declared compute group.
+ def tasks = sql_return_maparray """select * from tasks("type"="mv") where
MvName = '${mvName}'"""
+ assertTrue(tasks.size() > 0)
+ logger.info("mv tasks: ${tasks}")
+ assertEquals(cgName, tasks.get(0).ComputeGroup)
+
+ // ALTER keeps the same value space, including the DEFAULT rejection.
+ test {
+ sql """ALTER MATERIALIZED VIEW ${mvName} SET ('compute_group' =
'DEFAULT');"""
+ exception "reserved value"
+ }
+ test {
+ sql """ALTER MATERIALIZED VIEW ${mvName} SET ('compute_group' =
'cg_that_does_not_exist');"""
+ exception "not found"
+ }
+
+ // Re-declaring the same group is a no-op and must keep refreshing there.
+ sql """ALTER MATERIALIZED VIEW ${mvName} SET ('compute_group' =
'${cgName}');"""
+ sql """REFRESH MATERIALIZED VIEW ${mvName} AUTO;"""
+ waitingMTMVTaskFinishedByMvName(mvName)
+ def tasksAfterAlter = sql_return_maparray """select * from
tasks("type"="mv") where MvName = '${mvName}'"""
+ assertEquals(cgName, tasksAfterAlter.get(0).ComputeGroup)
+
+ sql """drop materialized view if exists ${mvName};"""
+ sql """drop table if exists `${tableName}`"""
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
