/*
 * 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.pinot.thirdeye.detection.components;

import java.util.concurrent.TimeUnit;
import org.apache.pinot.thirdeye.dataframe.DataFrame;
import org.apache.pinot.thirdeye.dataframe.util.MetricSlice;
import org.apache.pinot.thirdeye.datalayer.dto.MergedAnomalyResultDTO;
import org.apache.pinot.thirdeye.detection.InputDataFetcher;
import org.apache.pinot.thirdeye.detection.annotation.Components;
import org.apache.pinot.thirdeye.detection.annotation.DetectionTag;
import org.apache.pinot.thirdeye.detection.annotation.Param;
import org.apache.pinot.thirdeye.detection.annotation.PresentationOption;
import org.apache.pinot.thirdeye.detection.spec.ThresholdRuleFilterSpec;
import org.apache.pinot.thirdeye.detection.spi.components.AnomalyFilter;
import org.apache.pinot.thirdeye.detection.spi.model.InputData;
import org.apache.pinot.thirdeye.detection.spi.model.InputDataSpec;
import org.apache.pinot.thirdeye.rootcause.impl.MetricEntity;
import java.util.Collections;
import java.util.Map;
import org.joda.time.Interval;

import static org.apache.pinot.thirdeye.dataframe.util.DataFrameUtils.*;


/**
 * This threshold rule filter stage filters the anomalies if either the min or max thresholds do not pass.
 */
@Components(title = "Aggregate Threshold Filter", type = "THRESHOLD_RULE_FILTER", tags = {
    DetectionTag.RULE_FILTER}, description = "Threshold rule filter. filters the anomalies if either the min or max thresholds do not satisfied.")
public class ThresholdRuleAnomalyFilter implements AnomalyFilter<ThresholdRuleFilterSpec> {
  private double minValueHourly;
  private double maxValueHourly;
  private double minValueDaily;
  private double maxValueDaily;
  private double maxValue;
  private double minValue;
  @Override
  public boolean isQualified(MergedAnomalyResultDTO anomaly) {
    double currentValue = anomaly.getAvgCurrentVal();

    Interval anomalyInterval = new Interval(anomaly.getStartTime(), anomaly.getEndTime());
    double hourlyMultiplier = TimeUnit.HOURS.toMillis(1) / (double) anomalyInterval.toDurationMillis();
    double dailyMultiplier = TimeUnit.DAYS.toMillis(1) / (double) anomalyInterval.toDurationMillis();
    if (!Double.isNaN(this.minValue) && currentValue < this.minValue
        || !Double.isNaN(this.maxValue) && currentValue > this.maxValue) {
      return false;
    }
    if (!Double.isNaN(this.minValueHourly) && currentValue * hourlyMultiplier < this.minValueHourly) {
      return false;
    }
    if (!Double.isNaN(this.maxValueHourly) && currentValue * hourlyMultiplier > this.maxValueHourly) {
      return false;
    }
    if (!Double.isNaN(this.minValueDaily) && currentValue * dailyMultiplier < this.minValueDaily) {
      return false;
    }
    if (!Double.isNaN(this.maxValueDaily) && currentValue * dailyMultiplier > this.maxValueDaily) {
      return false;
    }
    return true;
  }

  @Override
  public void init(ThresholdRuleFilterSpec spec, InputDataFetcher dataFetcher) {
    this.minValueHourly = spec.getMinValueHourly();
    this.maxValueHourly = spec.getMaxValueHourly();
    this.minValueDaily = spec.getMinValueDaily();
    this.maxValueDaily = spec.getMaxValueDaily();
    this.maxValue = spec.getMaxValue();
    this.minValue = spec.getMinValue();
  }
}

@@ -880,81 +880,57 @@
 til.
-concurrent.TimeUnit;%0Aimport org.apache.pinot.thirdeye.dataframe.DataFrame
+Collections;%0Aimport java.util.concurrent.TimeUnit
 ;%0Aim
@@ -964,34 +964,34 @@
 eye.
-dataframe.util.MetricSlice
+constant.MetricAggFunction
 ;%0Aim
@@ -1059,16 +1059,80 @@
 ultDTO;%0A
+import org.apache.pinot.thirdeye.datalayer.dto.MetricConfigDTO;%0A
 import o
@@ -1322,143 +1322,8 @@
 ag;%0A
-import org.apache.pinot.thirdeye.detection.annotation.Param;%0Aimport org.apache.pinot.thirdeye.detection.annotation.PresentationOption;%0A
 impo
@@ -1468,72 +1468,8 @@
 er;%0A
-import org.apache.pinot.thirdeye.detection.spi.model.InputData;%0A
 impo
@@ -1605,156 +1605,30 @@
 ort 
-java.util.Collections;%0Aimport java.util.Map;%0Aimport org.joda.time.Interval;%0A%0Aimport static org.apache.pinot.thirdeye.dataframe.util.DataFrameUtils.*
+org.joda.time.Interval
 ;%0A%0A%0A
@@ -2256,16 +2256,57 @@
 nValue;%0A
+  private InputDataFetcher dataFetcher;%0A%0A
   @Overr
@@ -2368,24 +2368,261 @@
  anomaly) %7B%0A
+    MetricEntity me = MetricEntity.fromURN(anomaly.getMetricUrn());%0A    MetricConfigDTO metric = dataFetcher.fetchData(new InputDataSpec().withMetricIds(Collections.singleton(me.getId())))%0A        .getMetrics()%0A        .get(me.getId());%0A
     double c
@@ -2763,34 +2763,132 @@
     
-double hourlyMultiplier = 
+// apply multiplier if the metric is aggregated by SUM or COUNT%0A    double hourlyMultiplier =%0A        isAdditive(metric) ? (
 Time
@@ -2951,24 +2951,31 @@
 tionMillis()
+) : 1.0
 ;%0A    double
@@ -2992,17 +2992,47 @@
 iplier =
- 
+%0A        isAdditive(metric) ? (
 TimeUnit
@@ -3094,17 +3094,25 @@
 Millis()
-;
+) : 1.0;%0A
 %0A    if 
@@ -3806,16 +3806,245 @@
 e;%0A  %7D%0A%0A
+  private boolean isAdditive(MetricConfigDTO metric) %7B%0A    MetricAggFunction aggFunction = metric.getDefaultAggFunction();%0A    return aggFunction.equals(MetricAggFunction.SUM) %7C%7C aggFunction.equals(MetricAggFunction.COUNT);%0A  %7D%0A%0A
   @Overr
@@ -4412,14 +4412,50 @@
 alue();%0A
+    this.dataFetcher = dataFetcher;%0A
   %7D%0A%7D%0A

Add validations for the threshold rule anomaly filter (#5278)