feat: support discarding intermediate data when aggtask is updated (#818

)

server/agg/agg-core/src/main/java/io/holoinsight/server/agg/v1/core/conf/Extension.java

@@ -18,4 +18,10 @@ public class Extension {
    * Whether to enter debug mode
    */
   private boolean debug;
+
+  /**
+   * When an AggTask update is found, whether to discard the intermediate calculation results of the
+   * current cycle
+   */
+  private boolean discardWhenUpdate;
 }

server/agg/agg-executor/src/main/java/io/holoinsight/server/agg/v1/executor/executor/PartitionProcessor.java

@@ -126,7 +126,6 @@ private void processRecords(List<ConsumerRecord<AggTaskKey, AggProtos.AggTaskVal
       }
 
       AggTaskExecutor executor = getOrCreateAggTaskExecutor(key, latestAggTask);
-      executor.lastUsedAggTask = latestAggTask;
 
       RecursiveTask<Long> rt = new RecursiveTask<Long>() {
         @Override
@@ -165,8 +164,25 @@ protected Long compute() {
     state.updateMaxEventTimestamp(partitionMET);
   }
 
+  private void removeAggTaskExecutor(AggTaskKey key) {
+    aggTaskExecutors.remove(key);
+    state.getAggTaskStates().remove(key);
+  }
+
   private AggTaskExecutor getOrCreateAggTaskExecutor(AggTaskKey key, XAggTask aggTask) {
     AggTaskExecutor e = aggTaskExecutors.get(key);
+
+    // When an AggTask update is found, discard the intermediate calculation results of
+    // the current cycle. Doing so allows you to immediately update the day-level tasks instead of
+    // waiting until the next day to take effect. But the disadvantage is that the intermediate
+    // state will be lost and the data will be calculated from zero.
+    if (e != null && e.lastUsedAggTask != null
+        && aggTask.getInner().getExtension().isDiscardWhenUpdate()
+        && !e.lastUsedAggTask.hasSameVersion(aggTask)) {
+      removeAggTaskExecutor(key);
+      e = null;
+    }
+
     if (e == null) {
       AggTaskState s = new AggTaskState(key);
       long watermark = state.getMaxEventTimestamp() //
@@ -178,10 +194,10 @@ private AggTaskExecutor getOrCreateAggTaskExecutor(AggTaskKey key, XAggTask aggT
       s.setWatermark(watermark);
       e = new AggTaskExecutor(s, completenessService, output, aggMetaService);
       e.ignoredMinWatermark = watermark;
-      e.lastUsedAggTask = aggTask;
       aggTaskExecutors.put(key, e);
       state.put(s);
     }
+    e.lastUsedAggTask = aggTask;
     return e;
   }
 

server/agg/agg-executor/src/main/java/io/holoinsight/server/agg/v1/executor/executor/XAggTask.java

@@ -33,4 +33,8 @@ public XAggTask(AggTask inner, XSelect select, XWhere where) {
     this.select = select;
     this.where = where;
   }
+
+  public boolean hasSameVersion(XAggTask aggTask) {
+    return inner.getVersion() == aggTask.getInner().getVersion();
+  }
 }