[BugFix] Add task's defination into task run status

fe/fe-core/src/main/java/com/starrocks/catalog/system/information/TaskRunsSystemTable.java

@@ -47,6 +47,7 @@
 import org.apache.commons.lang3.NotImplementedException;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.util.Strings;
 import org.apache.thrift.meta_data.FieldValueMetaData;
 import org.apache.thrift.protocol.TType;
 
@@ -211,14 +212,18 @@ public static TGetTaskRunInfoResult query(TGetTasksParams params) {
             info.setState(status.getState().toString());
             info.setCatalog(status.getCatalogName());
             info.setDatabase(ClusterNamespace.getNameFromFullName(status.getDbName()));
-            try {
-                // NOTE: use task's definition to display task-run's definition here
-                Task task = taskManager.getTaskWithoutLock(taskName);
-                if (task != null) {
-                    info.setDefinition(task.getDefinition());
+            if (!Strings.isEmpty(status.getDefinition())) {
+                info.setDefinition(status.getDefinition());
+            } else {
+                try {
+                    // NOTE: use task's definition to display task-run's definition here
+                    Task task = taskManager.getTaskWithoutLock(taskName);
+                    if (task != null) {
+                        info.setDefinition(task.getDefinition());
+                    }
+                } catch (Exception e) {
+                    LOG.warn("Get taskName {} definition failed: {}", taskName, e);
                 }
-            } catch (Exception e) {
-                LOG.warn("Get taskName {} definition failed: {}", taskName, e);
             }
             info.setError_code(status.getErrorCode());
             info.setError_message(status.getErrorMessage());

fe/fe-core/src/main/java/com/starrocks/scheduler/TaskRun.java

@@ -340,6 +340,10 @@ public TaskRunStatus initStatus(String queryId, Long createTime) {
         status.setDbName(task.getDbName());
         status.setPostRun(task.getPostRun());
         status.setExpireTime(created + Config.task_runs_ttl_second * 1000L);
+        // NOTE: definition will cause a lot of repeats and cost a lot of metadata memory resources,
+        // since history task runs has been stored in sr's internal table, we can save it in the
+        // task run status.
+        status.setDefinition(task.getDefinition());
         status.getMvTaskRunExtraMessage().setExecuteOption(this.executeOption);
 
         LOG.info("init task status, task:{}, query_id:{}, create_time:{}", task.getName(), queryId, status.getCreateTime());

fe/fe-core/src/main/java/com/starrocks/scheduler/persist/TaskRunStatus.java

@@ -425,6 +425,13 @@ public boolean match(TGetTasksParams params) {
         return true;
     }
 
+    public String getDefinition() {
+        return definition;
+    }
+
+    public void setDefinition(String definition) {
+        this.definition = definition;
+    }
 
     public static TaskRunStatus read(DataInput in) throws IOException {
         String json = Text.readString(in);

fe/fe-core/src/test/java/com/starrocks/backup/RestoreJobMaterializedViewTest.java

@@ -510,7 +510,6 @@ public void testMVRestoreMVWithBaseTable2() {
         assertMVActiveEquals(MATERIALIZED_VIEW_NAME, true);
     }
 
-    @Test
     @Ignore
     public void testMVRestoreMVWithBaseTable3() {
         new MockUp<MetadataMgr>() {
@@ -530,7 +529,6 @@ public Table getTable(String catalogName, String dbName, String tblName) {
         assertMVActiveEquals(MATERIALIZED_VIEW_NAME, true);
     }
 
-    @Test
     @Ignore
     public void testMVRestoreMVWithBaseTable4() {
         new MockUp<MetadataMgr>() {

fe/fe-core/src/test/java/com/starrocks/scheduler/TaskManagerTest.java

@@ -799,4 +799,21 @@ public ConnectContext getRunCtx() {
         taskRunManager.killTaskRun(1L, true);
         Assert.assertEquals(0, taskRunScheduler.getRunningTaskCount());
     }
+
+    @Test
+    public void testTaskRunDefinition() {
+        Task task = new Task("test");
+        task.setDefinition("select 1");
+        long taskId = 1;
+        TaskRun taskRun = TaskRunBuilder
+                .newBuilder(task)
+                .setExecuteOption(DEFAULT_MERGE_OPTION)
+                .build();
+        long now = System.currentTimeMillis();
+        taskRun.setTaskId(taskId);
+        taskRun.initStatus("1", now + 10);
+        taskRun.getStatus().setPriority(0);
+        TaskRunStatus taskRunStatus = taskRun.getStatus();
+        Assert.assertEquals(taskRunStatus.getDefinition(), "select 1");
+    }
 }