This is an automated email from the ASF dual-hosted git repository.
morrySnow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 66dbb85fe3d [feature](fe) Show compute group for MTMV refresh task
(#63206)
66dbb85fe3d is described below
commit 66dbb85fe3deaf9069bc58f78446fe998ba5810b
Author: seawinde <[email protected]>
AuthorDate: Fri May 15 18:17:30 2026 +0800
[feature](fe) Show compute group for MTMV refresh task (#63206)
### What problem does this PR solve?
Problem Summary:
Async materialized view refresh task status does not show which compute
group runs the refresh task. This is needed for cloud deployments where
users inspect a single MTMV refresh task and want to know the compute
group used by that task.
Root cause: In `MTMVJobManager.refreshMTMV()`, manual refresh only
creates `MTMVTaskContext` with trigger mode, partitions, and
complete-refresh flag. The refresh execution later creates a new
background `ConnectContext` in `MTMVTask.exec()`, so the foreground
connection's selected compute group is not explicitly carried into the
task context.
| File | Change Description |
|------|-------------------|
| `MTMVJobManager.java` | Resolves the current connection's compute
group in cloud mode and stores it in `MTMVTaskContext` for manual
refresh. |
| `MTMVTaskContext.java` | Adds a serialized `computeGroup` field. |
| `MTMVTask.java` | Applies the task compute group to the background
refresh `ConnectContext`, records it, and exposes `ComputeGroup` in
`tasks("type" = "mv")`. |
| `MTMVTaskTest.java` | Covers schema, TVF output, context recording,
task context application, and old task JSON compatibility. |
| `MTMVJobManagerTest.java` | Covers passing the current compute group
from manual refresh into `MTMVTaskContext`. |
```mermaid
graph TD
A[User REFRESH MTMV] --> B[MTMVJobManager.refreshMTMV]
B --> C[Resolve current ConnectContext compute group]
C --> D[MTMVTaskContext.computeGroup]
D --> E[MTMVTask.exec creates background ConnectContext]
E --> F[Set cloud cluster on refresh ConnectContext]
F --> G[Execute refresh SQL]
F --> H[Expose ComputeGroup in tasks type mv]
```
Example:
After refreshing an async materialized view from compute group
`cg_analytics`:
```sql
REFRESH MATERIALIZED VIEW mv_sales;
```
Users can inspect the refresh task with `tasks("type" = "mv")` and read
the new `ComputeGroup` column:
```sql
SELECT
TaskId,
MvName,
Status,
RefreshMode,
LastQueryId,
ComputeGroup
FROM tasks("type" = "mv")
WHERE MvName = "mv_sales"
ORDER BY CreateTime DESC
LIMIT 1;
```
Example result:
```text
+--------+----------+---------+-------------+----------------------------------+--------------+
| TaskId | MvName | Status | RefreshMode | LastQueryId
| ComputeGroup |
+--------+----------+---------+-------------+----------------------------------+--------------+
| 12003 | mv_sales | SUCCESS | COMPLETE |
7f12c8d0f2a84b2b-9d8f2f4c1a5e6b7c | cg_analytics |
+--------+----------+---------+-------------+----------------------------------+--------------+
```
If the task has no recorded compute group, `ComputeGroup` is shown as
`\N`.
### Release note
Expose ComputeGroup in tasks("type" = "mv") for asynchronous
materialized view refresh tasks.
---
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 31 +++++++-
.../doris/job/extensions/mtmv/MTMVTaskContext.java | 12 ++-
.../java/org/apache/doris/mtmv/MTMVJobManager.java | 25 +++++-
.../org/apache/doris/mtmv/MTMVJobManagerTest.java | 88 ++++++++++++++++++++++
.../java/org/apache/doris/mtmv/MTMVTaskTest.java | 88 +++++++++++++++++++++-
5 files changed, 237 insertions(+), 7 deletions(-)
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 56a6ea56aec..7c9e65906ac 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
@@ -25,6 +25,7 @@ import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.ScalarType;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.cloud.qe.ComputeGroupException;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
@@ -70,6 +71,7 @@ import org.apache.doris.thrift.TRow;
import org.apache.doris.thrift.TStatusCode;
import org.apache.doris.thrift.TUniqueId;
+import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
@@ -114,7 +116,8 @@ public class MTMVTask extends AbstractTask {
new Column("NeedRefreshPartitions", ScalarType.createStringType()),
new Column("CompletedPartitions", ScalarType.createStringType()),
new Column("Progress", ScalarType.createStringType()),
- new Column("LastQueryId", ScalarType.createStringType()));
+ new Column("LastQueryId", ScalarType.createStringType()),
+ new Column("ComputeGroup", ScalarType.createStringType()));
public static final ImmutableMap<String, Integer> COLUMN_TO_INDEX;
@@ -152,6 +155,8 @@ public class MTMVTask extends AbstractTask {
MTMVTaskRefreshMode refreshMode;
@SerializedName("lastQueryId")
String lastQueryId;
+ @SerializedName("cg")
+ private String computeGroup;
private MTMV mtmv;
private MTMVRelation relation;
@@ -322,6 +327,8 @@ public class MTMVTask extends AbstractTask {
Map<TableIf, String> tableWithPartKey)
throws Exception {
ConnectContext ctx = MTMVPlanUtil.createMTMVContext(mtmv,
MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK);
+ setComputeGroup(ctx);
+ recordComputeGroup(ctx);
StatementContext statementContext = new StatementContext();
for (Entry<MvccTableInfo, MvccSnapshot> entry : snapshots.entrySet()) {
statementContext.setSnapshot(entry.getKey(), entry.getValue());
@@ -354,6 +361,26 @@ public class MTMVTask extends AbstractTask {
}
}
+ private void setComputeGroup(ConnectContext ctx) {
+ String taskComputeGroup = taskContext.getComputeGroup();
+ if (Config.isCloudMode() && !Strings.isNullOrEmpty(taskComputeGroup)) {
+ ctx.setCloudCluster(taskComputeGroup);
+ }
+ }
+
+ private void recordComputeGroup(ConnectContext ctx) {
+ if (!Config.isCloudMode()) {
+ computeGroup = FeConstants.null_string;
+ return;
+ }
+ try {
+ computeGroup = ctx.getCloudCluster(false);
+ } catch (ComputeGroupException e) {
+ computeGroup = FeConstants.null_string;
+ LOG.warn("failed to resolve compute group for mtmv task, taskId:
{}", getTaskId(), e);
+ }
+ }
+
private String getDummyStmt(Set<String> refreshPartitionNames) {
String mvName = mtmv.getName();
DatabaseIf database = mtmv.getDatabase();
@@ -532,6 +559,8 @@ public class MTMVTask extends AbstractTask {
new TCell().setStringVal(getProgress()));
trow.addToColumnValue(
new TCell().setStringVal(lastQueryId));
+ trow.addToColumnValue(new TCell().setStringVal(
+ computeGroup == null || computeGroup.isEmpty() ?
FeConstants.null_string : computeGroup));
return trow;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTaskContext.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTaskContext.java
index 1c53f5dd301..1d81a5d6a44 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTaskContext.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTaskContext.java
@@ -34,14 +34,19 @@ public class MTMVTaskContext {
@SerializedName(value = "isComplete")
private boolean isComplete;
+ @SerializedName(value = "computeGroup")
+ private String computeGroup;
+
public MTMVTaskContext(MTMVTaskTriggerMode triggerMode) {
this.triggerMode = triggerMode;
}
- public MTMVTaskContext(MTMVTaskTriggerMode triggerMode, List<String>
partitions, boolean isComplete) {
+ public MTMVTaskContext(MTMVTaskTriggerMode triggerMode, List<String>
partitions, boolean isComplete,
+ String computeGroup) {
this.triggerMode = triggerMode;
this.partitions = partitions;
this.isComplete = isComplete;
+ this.computeGroup = computeGroup;
}
public List<String> getPartitions() {
@@ -56,12 +61,17 @@ public class MTMVTaskContext {
return isComplete;
}
+ public String getComputeGroup() {
+ return computeGroup;
+ }
+
@Override
public String toString() {
return "MTMVTaskContext{"
+ "triggerMode=" + triggerMode
+ ", partitions=" + partitions
+ ", isComplete=" + isComplete
+ + ", computeGroup=" + computeGroup
+ '}';
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVJobManager.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVJobManager.java
index 7bee416a849..67a5bbe4db5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVJobManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVJobManager.java
@@ -24,6 +24,8 @@ import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.TableIf.TableType;
import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.cloud.qe.ComputeGroupException;
+import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.util.TimeUtils;
@@ -42,6 +44,7 @@ import
org.apache.doris.nereids.trees.plans.commands.info.CancelMTMVTaskInfo;
import org.apache.doris.nereids.trees.plans.commands.info.PauseMTMVInfo;
import org.apache.doris.nereids.trees.plans.commands.info.RefreshMTMVInfo;
import org.apache.doris.nereids.trees.plans.commands.info.ResumeMTMVInfo;
+import org.apache.doris.qe.ConnectContext;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;
@@ -64,7 +67,7 @@ public class MTMVJobManager implements MTMVHookService {
if (!mtmv.getRefreshInfo().getBuildMode().equals(BuildMode.IMMEDIATE))
{
return;
}
- MTMVTaskContext mtmvTaskContext = new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM, null, true);
+ MTMVTaskContext mtmvTaskContext = new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM, null, true, null);
try {
Env.getCurrentEnv().getJobManager().triggerJob(mtmv.getId(),
mtmvTaskContext);
} catch (JobException e) {
@@ -155,10 +158,26 @@ public class MTMVJobManager implements MTMVHookService {
public void refreshMTMV(RefreshMTMVInfo info) throws DdlException,
MetaNotFoundException, JobException {
MTMVJob job = getJobByTableNameInfo(info.getMvName());
MTMVTaskContext mtmvTaskContext = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, info.getPartitions(),
- info.isComplete());
+ info.isComplete(), getCurrentComputeGroup());
Env.getCurrentEnv().getJobManager().triggerJob(job.getJobId(),
mtmvTaskContext);
}
+ private String getCurrentComputeGroup() {
+ if (!Config.isCloudMode()) {
+ return null;
+ }
+ ConnectContext ctx = ConnectContext.get();
+ if (ctx == null) {
+ return null;
+ }
+ try {
+ return ctx.getCloudCluster(false);
+ } catch (ComputeGroupException e) {
+ LOG.warn("failed to resolve compute group for refresh mtmv", e);
+ return null;
+ }
+ }
+
@Override
public void refreshComplete(MTMV mtmv, MTMVRelation relation, MTMVTask
task) {
@@ -202,7 +221,7 @@ public class MTMVJobManager implements MTMVHookService {
return;
}
MTMVTaskContext mtmvTaskContext = new
MTMVTaskContext(MTMVTaskTriggerMode.COMMIT, Lists.newArrayList(),
- false);
+ false, null);
Env.getCurrentEnv().getJobManager().triggerJob(job.getJobId(),
mtmvTaskContext);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVJobManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVJobManagerTest.java
new file mode 100644
index 00000000000..b97dc6b4d99
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVJobManagerTest.java
@@ -0,0 +1,88 @@
+// 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.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.TableIf.TableType;
+import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.common.Config;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.job.extensions.mtmv.MTMVJob;
+import org.apache.doris.job.extensions.mtmv.MTMVTask.MTMVTaskTriggerMode;
+import org.apache.doris.job.extensions.mtmv.MTMVTaskContext;
+import org.apache.doris.job.manager.JobManager;
+import org.apache.doris.nereids.trees.plans.commands.info.RefreshMTMVInfo;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.collect.Lists;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+public class MTMVJobManagerTest {
+
+ @Test
+ public void testRefreshMTMVPassesCurrentComputeGroupToTaskContext() throws
Exception {
+ String originCloudUniqueId = Config.cloud_unique_id;
+ ConnectContext previousContext = ConnectContext.get();
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ Config.cloud_unique_id = "test_cloud";
+ Env env = Mockito.mock(Env.class);
+ InternalCatalog internalCatalog =
Mockito.mock(InternalCatalog.class);
+ Database db = Mockito.mock(Database.class);
+ MTMV mtmv = Mockito.mock(MTMV.class);
+ MTMVJob job = Mockito.mock(MTMVJob.class);
+ JobManager jobManager = Mockito.mock(JobManager.class);
+
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+
mockedEnv.when(Env::getCurrentInternalCatalog).thenReturn(internalCatalog);
+
Mockito.when(internalCatalog.getDbOrDdlException("db1")).thenReturn(db);
+ Mockito.when(db.getTableOrMetaException(Mockito.eq("mv1"),
Mockito.eq(TableType.MATERIALIZED_VIEW)))
+ .thenReturn(mtmv);
+ Mockito.when(env.getJobManager()).thenReturn(jobManager);
+ Mockito.when(jobManager.getJob(mtmv.getId())).thenReturn(job);
+ Mockito.when(job.getJobId()).thenReturn(100L);
+
+ ConnectContext ctx = new ConnectContext();
+ ctx.setCloudCluster("cg1");
+ ctx.setThreadLocalInfo();
+
+ RefreshMTMVInfo info = new RefreshMTMVInfo(new
TableNameInfo("db1", "mv1"),
+ Lists.newArrayList("p1"), false);
+ new MTMVJobManager().refreshMTMV(info);
+
+ ArgumentCaptor<MTMVTaskContext> captor =
ArgumentCaptor.forClass(MTMVTaskContext.class);
+ Mockito.verify(jobManager).triggerJob(Mockito.eq(100L),
captor.capture());
+ MTMVTaskContext taskContext = captor.getValue();
+ Assert.assertEquals(MTMVTaskTriggerMode.MANUAL,
taskContext.getTriggerMode());
+ Assert.assertEquals(Lists.newArrayList("p1"),
taskContext.getPartitions());
+ Assert.assertFalse(taskContext.isComplete());
+ Assert.assertEquals("cg1", taskContext.getComputeGroup());
+ } finally {
+ Config.cloud_unique_id = originCloudUniqueId;
+ ConnectContext.remove();
+ if (previousContext != null) {
+ previousContext.setThreadLocalInfo();
+ }
+ }
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
index 4e4b51a531f..505f210fc2f 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
@@ -17,15 +17,22 @@
package org.apache.doris.mtmv;
+import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.MTMV;
import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
+import org.apache.doris.common.FeConstants;
import org.apache.doris.common.MetaNotFoundException;
+import org.apache.doris.common.jmockit.Deencapsulation;
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.mtmv.MTMVPartitionInfo.MTMVPartitionType;
import org.apache.doris.mtmv.MTMVRefreshEnum.RefreshMethod;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.thrift.TRow;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -52,6 +59,7 @@ public class MTMVTaskTest {
private MTMVRefreshInfo mtmvRefreshInfo =
Mockito.mock(MTMVRefreshInfo.class);
private MockedStatic<MTMVUtil> mtmvUtilStatic;
private MockedStatic<MTMVPartitionUtil> mtmvPartitionUtilStatic;
+ private static final String COMPUTE_GROUP = "ComputeGroup";
@Before
public void setUp()
@@ -87,7 +95,7 @@ public class MTMVTaskTest {
@Test
public void testCalculateNeedRefreshPartitionsManualComplete() throws
AnalysisException {
- MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, null, true);
+ MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, null, true, null);
MTMVTask task = new MTMVTask(mtmv, relation, context);
List<String> result = task.calculateNeedRefreshPartitions(null);
Assert.assertEquals(allPartitionNames, result);
@@ -95,7 +103,8 @@ public class MTMVTaskTest {
@Test
public void testCalculateNeedRefreshPartitionsManualPartitions() throws
AnalysisException {
- MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, Lists.newArrayList(poneName),
false);
+ MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, Lists.newArrayList(poneName),
+ false, null);
MTMVTask task = new MTMVTask(mtmv, relation, context);
List<String> result = task.calculateNeedRefreshPartitions(null);
Assert.assertEquals(Lists.newArrayList(poneName), result);
@@ -139,4 +148,79 @@ public class MTMVTaskTest {
List<String> result = task.calculateNeedRefreshPartitions(null);
Assert.assertEquals(Lists.newArrayList(ptwoName), result);
}
+
+ @Test
+ public void testTaskSchemaContainsComputeGroup() {
+ Column lastColumn = MTMVTask.SCHEMA.get(MTMVTask.SCHEMA.size() - 1);
+ Assert.assertEquals(COMPUTE_GROUP, lastColumn.getName());
+ Assert.assertEquals(MTMVTask.SCHEMA.size() - 1,
+
MTMVTask.COLUMN_TO_INDEX.get(COMPUTE_GROUP.toLowerCase()).intValue());
+ }
+
+ @Test
+ public void testGetTvfInfoReturnsComputeGroup() {
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ Deencapsulation.setField(task, "computeGroup", "cg1");
+
+ TRow row = task.getTvfInfo("job1");
+
+ Assert.assertEquals("cg1", row.getColumnValue()
+
.get(MTMVTask.COLUMN_TO_INDEX.get(COMPUTE_GROUP.toLowerCase())).getStringVal());
+ }
+
+ @Test
+ public void testRecordComputeGroupFromContext() {
+ String originCloudUniqueId = Config.cloud_unique_id;
+ try {
+ Config.cloud_unique_id = "test_cloud";
+ ConnectContext ctx = new ConnectContext();
+ ctx.setCloudCluster("cg1");
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+
+ Deencapsulation.invoke(task, "recordComputeGroup", ctx);
+ TRow row = task.getTvfInfo("job1");
+
+ Assert.assertEquals("cg1", row.getColumnValue()
+
.get(MTMVTask.COLUMN_TO_INDEX.get(COMPUTE_GROUP.toLowerCase())).getStringVal());
+ } finally {
+ Config.cloud_unique_id = originCloudUniqueId;
+ }
+ }
+
+ @Test
+ public void testSetComputeGroupFromTaskContext() {
+ String originCloudUniqueId = Config.cloud_unique_id;
+ try {
+ Config.cloud_unique_id = "test_cloud";
+ ConnectContext ctx = new ConnectContext();
+ MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, null, true, "cg1");
+ MTMVTask task = new MTMVTask(mtmv, relation, context);
+
+ Deencapsulation.invoke(task, "setComputeGroup", ctx);
+
+ Assert.assertEquals("cg1",
ctx.getSessionVariable().getCloudCluster());
+ } finally {
+ Config.cloud_unique_id = originCloudUniqueId;
+ }
+ }
+
+ @Test
+ public void testGetTvfInfoReturnsNullStringForMissingComputeGroup() {
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+
+ TRow row = task.getTvfInfo("job1");
+
+ Assert.assertEquals(FeConstants.null_string, row.getColumnValue()
+
.get(MTMVTask.COLUMN_TO_INDEX.get(COMPUTE_GROUP.toLowerCase())).getStringVal());
+ }
+
+ @Test
+ public void testDeserializeOldTaskWithoutComputeGroup() {
+ MTMVTask task = GsonUtils.GSON.fromJson("{\"di\":1,\"mi\":2}",
MTMVTask.class);
+
+ TRow row = task.getTvfInfo("job1");
+
+ Assert.assertEquals(FeConstants.null_string, row.getColumnValue()
+
.get(MTMVTask.COLUMN_TO_INDEX.get(COMPUTE_GROUP.toLowerCase())).getStringVal());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
