You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2020/07/30 20:35:13 UTC
[GitHub] [incubator-pinot] jihaozh opened a new pull request #5778: [TE] The endpoint for searching anomalies and pagination
jihaozh opened a new pull request #5778:
URL: https://github.com/apache/incubator-pinot/pull/5778
This PR creates the endpoints for searching and paginate the anomalies. The endpoint is built for the anomaly page v3. It brings the response time from ~30s to around 300ms. We should see a ~100x speedup of the page load time.
The current Anomaly list page is loading slowly because of calling legacy endpoints. Although the old endpoint works ok when the scale is small, it responses very slowly when handling anomalies on a large scale. (above 100k+ anomalies)
This endpoint pushes the filtering and pagination of the anomalies into backend and database servers instead of doing it in the UI.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] jihaozh commented on a change in pull request #5778: [TE] The endpoint for searching anomalies and pagination
Posted by GitBox <gi...@apache.org>.
jihaozh commented on a change in pull request #5778:
URL: https://github.com/apache/incubator-pinot/pull/5778#discussion_r465261202
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearchResource.java
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.dashboard.resources.v2.anomalies;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import java.util.List;
+import javax.ws.rs.DefaultValue;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.pinot.thirdeye.api.Constants;
+
+
+/**
+ * The type Anomaly search resource.
+ */
+@Path(value = "/anomaly-search")
Review comment:
I'd prefer to keep them separate and eventually we will want to deprecate the old file.
Reasons:
- the input for the endpoints is quite different, the new one needs to support filter by multiple datasets, metrics, while the old one can only handle a single filter in each field.
- the response formatting for this one is quite different. In the new one, it needs to add the pagination information, the anomaly result is on another level in the blob.
Therefore just updating the old one is not going to be backward compatible.
Also, in the old one, in the case of multiple filters, the filters are applied in memory instead of pushing to the database, so it's not efficient when handling a large number of anomalies.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] bxji commented on a change in pull request #5778: [TE] The endpoint for searching anomalies and pagination
Posted by GitBox <gi...@apache.org>.
bxji commented on a change in pull request #5778:
URL: https://github.com/apache/incubator-pinot/pull/5778#discussion_r463314454
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearchResource.java
##########
@@ -0,0 +1,61 @@
+package org.apache.pinot.thirdeye.dashboard.resources.v2.anomalies;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import java.util.List;
+import javax.ws.rs.DefaultValue;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
Review comment:
add header
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearchFilter.java
##########
@@ -0,0 +1,132 @@
+package org.apache.pinot.thirdeye.dashboard.resources.v2.anomalies;
+
+import java.util.Collections;
+import java.util.List;
+
+
Review comment:
add header
##########
File path: thirdeye/thirdeye-pinot/src/test/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearcherTest.java
##########
@@ -0,0 +1,68 @@
+package org.apache.pinot.thirdeye.dashboard.resources.v2.anomalies;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
Review comment:
header
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearcher.java
##########
@@ -0,0 +1,142 @@
+package org.apache.pinot.thirdeye.dashboard.resources.v2.anomalies;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
Review comment:
add header
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearcher.java
##########
@@ -0,0 +1,142 @@
+package org.apache.pinot.thirdeye.dashboard.resources.v2.anomalies;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.pinot.thirdeye.constant.AnomalyFeedbackType;
+import org.apache.pinot.thirdeye.datalayer.bao.DetectionAlertConfigManager;
+import org.apache.pinot.thirdeye.datalayer.bao.DetectionConfigManager;
+import org.apache.pinot.thirdeye.datalayer.bao.MergedAnomalyResultManager;
+import org.apache.pinot.thirdeye.datalayer.dto.AbstractDTO;
+import org.apache.pinot.thirdeye.datalayer.dto.MergedAnomalyResultDTO;
+import org.apache.pinot.thirdeye.datalayer.pojo.DetectionConfigBean;
+import org.apache.pinot.thirdeye.datalayer.util.Predicate;
+import org.apache.pinot.thirdeye.datasource.DAORegistry;
+
+import static org.apache.pinot.thirdeye.constant.AnomalyFeedbackType.*;
+
+
+/**
+ * The type Anomaly searcher.
+ */
+public class AnomalySearcher {
+ private final MergedAnomalyResultManager anomalyDAO;
+ private final DetectionConfigManager detectionConfigDAO;
+ private final DetectionAlertConfigManager detectionAlertConfigDAO;
+
+ /**
+ * Instantiates a new Anomaly searcher.
+ */
+ public AnomalySearcher() {
+ this.anomalyDAO = DAORegistry.getInstance().getMergedAnomalyResultDAO();
+ this.detectionConfigDAO = DAORegistry.getInstance().getDetectionConfigManager();
+ this.detectionAlertConfigDAO = DAORegistry.getInstance().getDetectionAlertConfigManager();
+ }
+
+ /**
+ * Search and retrieve all the anomalies matching to the search filter and limits.
+ *
+ * @param searchFilter the search filter
+ * @param limit the limit
+ * @param offset the offset
+ * @return the result
+ */
+ public Map<String, Object> search(AnomalySearchFilter searchFilter, int limit, int offset) {
+ Predicate predicate = Predicate.EQ("child", false);
+ if (searchFilter.getStartTime() != null) {
+ predicate = Predicate.AND(predicate, Predicate.LT("startTime", searchFilter.getEndTime()));
+ }
+ if (searchFilter.getEndTime() != null) {
+ predicate = Predicate.AND(predicate, Predicate.GT("endTime", searchFilter.getStartTime()));
+ }
+ // search by detections or subscription groups
+ Set<Long> detectionConfigIds = new HashSet<>();
+ Set<Long> subscribedDetectionConfigIds = new HashSet<>();
+ if (!searchFilter.getDetectionNames().isEmpty()) {
+ detectionConfigIds =
+ this.detectionConfigDAO.findByPredicate(Predicate.IN("name", searchFilter.getDetectionNames().toArray()))
+ .stream()
+ .map(DetectionConfigBean::getId)
+ .collect(Collectors.toSet());
+ }
+ if (!searchFilter.getSubscriptionGroups().isEmpty()) {
+ subscribedDetectionConfigIds = this.detectionAlertConfigDAO.findByPredicate(
+ Predicate.IN("name", searchFilter.getSubscriptionGroups().toArray()))
+ .stream()
+ .map(detectionAlertConfigDTO -> detectionAlertConfigDTO.getVectorClocks().keySet())
+ .flatMap(Collection::stream)
+ .collect(Collectors.toSet());
+ }
+ if (!searchFilter.getDetectionNames().isEmpty() && !searchFilter.getSubscriptionGroups().isEmpty()) {
+ // intersect the detection config ids if searching by both
+ detectionConfigIds.retainAll(subscribedDetectionConfigIds);
+ } else {
+ detectionConfigIds.addAll(subscribedDetectionConfigIds);
+ }
+ if (!searchFilter.getDetectionNames().isEmpty() || !searchFilter.getSubscriptionGroups().isEmpty()) {
+ // add the predicate using detection config id
+ if (detectionConfigIds.isEmpty()) {
+ // if detection not found, return empty result
+ return ImmutableMap.of("count", 0, "limit", limit, "offset", offset, "elements", Collections.emptyList());
+ }
+ predicate = Predicate.AND(predicate, Predicate.IN("detectionConfigId", detectionConfigIds.toArray()));
+ }
+
+ // search by datasets
+ if (!searchFilter.getDatasets().isEmpty()) {
+ List<Predicate> datasetPredicates = new ArrayList<>();
+ for (String dataset : searchFilter.getDatasets()) {
+ datasetPredicates.add(Predicate.LIKE("collection", "%" + dataset + "%"));
+ }
+ predicate = Predicate.AND(predicate, Predicate.OR(datasetPredicates.toArray(new Predicate[0])));
+ }
+ // search by metrics
+ if (!searchFilter.getMetrics().isEmpty()) {
+ predicate = Predicate.AND(predicate, Predicate.IN("metric", searchFilter.getMetrics().toArray()));
+ }
+ // search by ids
+ if (!searchFilter.getAnomalyIds().isEmpty()) {
+ predicate = Predicate.AND(predicate, Predicate.IN("baseId", searchFilter.getAnomalyIds().toArray()));
+ }
+
+ long count;
+ List<MergedAnomalyResultDTO> results;
+ if (searchFilter.getFeedbacks().isEmpty()) {
+ List<Long> anomalyIds = this.anomalyDAO.findIdsByPredicate(predicate)
+ .stream()
+ .sorted(Comparator.reverseOrder())
+ .collect(Collectors.toList());
+ count = anomalyIds.size();
+ results = anomalyIds.isEmpty() ? Collections.emptyList()
+ : this.anomalyDAO.findByIds(paginateResults(anomalyIds, offset, limit));
+ } else {
+ // filter by feedback types if requested
+ List<MergedAnomalyResultDTO> anomalies = this.anomalyDAO.findByPredicate(predicate);
+ Set<AnomalyFeedbackType> feedbackFilters =
+ searchFilter.getFeedbacks().stream().map(AnomalyFeedbackType::valueOf).collect(Collectors.toSet());
+ results = anomalies.stream()
+ .filter(anomaly -> (anomaly.getFeedback() == null && feedbackFilters.contains(NO_FEEDBACK)) || (
+ anomaly.getFeedback() != null && feedbackFilters.contains(anomaly.getFeedback().getFeedbackType())))
+ .sorted(Comparator.comparingLong(AbstractDTO::getId).reversed())
+ .collect(Collectors.toList());
+ count = results.size();
+ results = paginateResults(results, offset, limit);
+ }
+ return ImmutableMap.of("count", count, "limit", limit, "offset", offset, "elements", results);
+ }
+
+ private <T> List<T> paginateResults(List<T> list, int offset, int limit) {
Review comment:
Just want to check my understanding; I see that we are starting with offset of 0. So my understanding is that if we have limit 25, the anomalies returned should be 0-24?
If so, since we are starting with offset of 0, do we need >= as the equality check here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] jihaozh merged pull request #5778: [TE] The endpoint for searching anomalies and pagination
Posted by GitBox <gi...@apache.org>.
jihaozh merged pull request #5778:
URL: https://github.com/apache/incubator-pinot/pull/5778
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request #5778: [TE] The endpoint for searching anomalies and pagination
Posted by GitBox <gi...@apache.org>.
akshayrai commented on a change in pull request #5778:
URL: https://github.com/apache/incubator-pinot/pull/5778#discussion_r465204182
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearchResource.java
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.dashboard.resources.v2.anomalies;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import java.util.List;
+import javax.ws.rs.DefaultValue;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.pinot.thirdeye.api.Constants;
+
+
+/**
+ * The type Anomaly search resource.
+ */
+@Path(value = "/anomaly-search")
+@Produces(MediaType.APPLICATION_JSON)
+@Api(tags = {Constants.DETECTION_TAG})
+public class AnomalySearchResource {
+
+ private final AnomalySearcher anomalySearcher;
+
+ /**
+ * Instantiates a new Anomaly search resource.
+ */
+ public AnomalySearchResource() {
+ this.anomalySearcher = new AnomalySearcher();
+ }
+
+ /**
+ * Search and paginate the anomalies according to the parameters.
+ *
+ * @param limit the limit
+ * @param offset the offset
+ * @param startTime the start time
+ * @param endTime the end time
+ * @param feedbacks the feedback types, e.g. ANOMALY, NOT_ANOMALY
+ * @param subscriptionGroups the subscription groups
+ * @param detectionNames the detection names
+ * @param metrics the metrics
+ * @param datasets the datasets
+ * @param anomalyIds the anomaly ids
+ * @return the response
Review comment:
You might want to describe these fields better or if they are intuitive enough I would usually not make an entry at all.
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearchResource.java
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.dashboard.resources.v2.anomalies;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import java.util.List;
+import javax.ws.rs.DefaultValue;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.pinot.thirdeye.api.Constants;
+
+
+/**
+ * The type Anomaly search resource.
+ */
+@Path(value = "/anomaly-search")
Review comment:
How about merging and enriching the existing query anomalies endpoint in `UserDashboardResource`?
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearcher.java
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.dashboard.resources.v2.anomalies;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.pinot.thirdeye.constant.AnomalyFeedbackType;
+import org.apache.pinot.thirdeye.datalayer.bao.DetectionAlertConfigManager;
+import org.apache.pinot.thirdeye.datalayer.bao.DetectionConfigManager;
+import org.apache.pinot.thirdeye.datalayer.bao.MergedAnomalyResultManager;
+import org.apache.pinot.thirdeye.datalayer.dto.AbstractDTO;
+import org.apache.pinot.thirdeye.datalayer.dto.MergedAnomalyResultDTO;
+import org.apache.pinot.thirdeye.datalayer.pojo.DetectionConfigBean;
+import org.apache.pinot.thirdeye.datalayer.util.Predicate;
+import org.apache.pinot.thirdeye.datasource.DAORegistry;
+
+import static org.apache.pinot.thirdeye.constant.AnomalyFeedbackType.*;
+
+
+/**
+ * The type Anomaly searcher.
+ */
+public class AnomalySearcher {
+ private final MergedAnomalyResultManager anomalyDAO;
+ private final DetectionConfigManager detectionConfigDAO;
+ private final DetectionAlertConfigManager detectionAlertConfigDAO;
+
+ /**
+ * Instantiates a new Anomaly searcher.
+ */
+ public AnomalySearcher() {
+ this.anomalyDAO = DAORegistry.getInstance().getMergedAnomalyResultDAO();
+ this.detectionConfigDAO = DAORegistry.getInstance().getDetectionConfigManager();
+ this.detectionAlertConfigDAO = DAORegistry.getInstance().getDetectionAlertConfigManager();
+ }
+
+ /**
+ * Search and retrieve all the anomalies matching to the search filter and limits.
+ *
+ * @param searchFilter the search filter
+ * @param limit the limit
+ * @param offset the offset
+ * @return the result
+ */
+ public Map<String, Object> search(AnomalySearchFilter searchFilter, int limit, int offset) {
+ Predicate predicate = Predicate.EQ("child", false);
+ if (searchFilter.getStartTime() != null) {
+ predicate = Predicate.AND(predicate, Predicate.LT("startTime", searchFilter.getEndTime()));
+ }
+ if (searchFilter.getEndTime() != null) {
+ predicate = Predicate.AND(predicate, Predicate.GT("endTime", searchFilter.getStartTime()));
+ }
+ // search by detections or subscription groups
+ Set<Long> detectionConfigIds = new HashSet<>();
+ Set<Long> subscribedDetectionConfigIds = new HashSet<>();
+ if (!searchFilter.getDetectionNames().isEmpty()) {
+ detectionConfigIds =
+ this.detectionConfigDAO.findByPredicate(Predicate.IN("name", searchFilter.getDetectionNames().toArray()))
+ .stream()
+ .map(DetectionConfigBean::getId)
+ .collect(Collectors.toSet());
+ }
+ if (!searchFilter.getSubscriptionGroups().isEmpty()) {
+ subscribedDetectionConfigIds = this.detectionAlertConfigDAO.findByPredicate(
+ Predicate.IN("name", searchFilter.getSubscriptionGroups().toArray()))
+ .stream()
+ .map(detectionAlertConfigDTO -> detectionAlertConfigDTO.getVectorClocks().keySet())
+ .flatMap(Collection::stream)
+ .collect(Collectors.toSet());
+ }
+ if (!searchFilter.getDetectionNames().isEmpty() && !searchFilter.getSubscriptionGroups().isEmpty()) {
+ // intersect the detection config ids if searching by both
+ detectionConfigIds.retainAll(subscribedDetectionConfigIds);
+ } else {
+ detectionConfigIds.addAll(subscribedDetectionConfigIds);
+ }
+ if (!searchFilter.getDetectionNames().isEmpty() || !searchFilter.getSubscriptionGroups().isEmpty()) {
+ // add the predicate using detection config id
+ if (detectionConfigIds.isEmpty()) {
+ // if detection not found, return empty result
+ return ImmutableMap.of("count", 0, "limit", limit, "offset", offset, "elements", Collections.emptyList());
+ }
+ predicate = Predicate.AND(predicate, Predicate.IN("detectionConfigId", detectionConfigIds.toArray()));
+ }
+
+ // search by datasets
+ if (!searchFilter.getDatasets().isEmpty()) {
+ List<Predicate> datasetPredicates = new ArrayList<>();
+ for (String dataset : searchFilter.getDatasets()) {
+ datasetPredicates.add(Predicate.LIKE("collection", "%" + dataset + "%"));
+ }
+ predicate = Predicate.AND(predicate, Predicate.OR(datasetPredicates.toArray(new Predicate[0])));
+ }
+ // search by metrics
+ if (!searchFilter.getMetrics().isEmpty()) {
+ predicate = Predicate.AND(predicate, Predicate.IN("metric", searchFilter.getMetrics().toArray()));
+ }
+ // search by ids
+ if (!searchFilter.getAnomalyIds().isEmpty()) {
+ predicate = Predicate.AND(predicate, Predicate.IN("baseId", searchFilter.getAnomalyIds().toArray()));
+ }
+
+ long count;
+ List<MergedAnomalyResultDTO> results;
+ if (searchFilter.getFeedbacks().isEmpty()) {
+ List<Long> anomalyIds = this.anomalyDAO.findIdsByPredicate(predicate)
+ .stream()
+ .sorted(Comparator.reverseOrder())
+ .collect(Collectors.toList());
+ count = anomalyIds.size();
+ results = anomalyIds.isEmpty() ? Collections.emptyList()
+ : this.anomalyDAO.findByIds(paginateResults(anomalyIds, offset, limit));
+ } else {
+ // filter by feedback types if requested
+ List<MergedAnomalyResultDTO> anomalies = this.anomalyDAO.findByPredicate(predicate);
+ Set<AnomalyFeedbackType> feedbackFilters =
+ searchFilter.getFeedbacks().stream().map(AnomalyFeedbackType::valueOf).collect(Collectors.toSet());
+ results = anomalies.stream()
+ .filter(anomaly -> (anomaly.getFeedback() == null && feedbackFilters.contains(NO_FEEDBACK)) || (
+ anomaly.getFeedback() != null && feedbackFilters.contains(anomaly.getFeedback().getFeedbackType())))
+ .sorted(Comparator.comparingLong(AbstractDTO::getId).reversed())
+ .collect(Collectors.toList());
+ count = results.size();
+ results = paginateResults(results, offset, limit);
+ }
+ return ImmutableMap.of("count", count, "limit", limit, "offset", offset, "elements", results);
+ }
+
+ private <T> List<T> paginateResults(List<T> list, int offset, int limit) {
Review comment:
This is generic enough to move out into a utility?
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearcher.java
##########
@@ -0,0 +1,142 @@
+package org.apache.pinot.thirdeye.dashboard.resources.v2.anomalies;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.pinot.thirdeye.constant.AnomalyFeedbackType;
+import org.apache.pinot.thirdeye.datalayer.bao.DetectionAlertConfigManager;
+import org.apache.pinot.thirdeye.datalayer.bao.DetectionConfigManager;
+import org.apache.pinot.thirdeye.datalayer.bao.MergedAnomalyResultManager;
+import org.apache.pinot.thirdeye.datalayer.dto.AbstractDTO;
+import org.apache.pinot.thirdeye.datalayer.dto.MergedAnomalyResultDTO;
+import org.apache.pinot.thirdeye.datalayer.pojo.DetectionConfigBean;
+import org.apache.pinot.thirdeye.datalayer.util.Predicate;
+import org.apache.pinot.thirdeye.datasource.DAORegistry;
+
+import static org.apache.pinot.thirdeye.constant.AnomalyFeedbackType.*;
+
+
+/**
+ * The type Anomaly searcher.
+ */
+public class AnomalySearcher {
+ private final MergedAnomalyResultManager anomalyDAO;
+ private final DetectionConfigManager detectionConfigDAO;
+ private final DetectionAlertConfigManager detectionAlertConfigDAO;
+
+ /**
+ * Instantiates a new Anomaly searcher.
+ */
+ public AnomalySearcher() {
+ this.anomalyDAO = DAORegistry.getInstance().getMergedAnomalyResultDAO();
+ this.detectionConfigDAO = DAORegistry.getInstance().getDetectionConfigManager();
+ this.detectionAlertConfigDAO = DAORegistry.getInstance().getDetectionAlertConfigManager();
+ }
+
+ /**
+ * Search and retrieve all the anomalies matching to the search filter and limits.
+ *
+ * @param searchFilter the search filter
+ * @param limit the limit
+ * @param offset the offset
+ * @return the result
+ */
+ public Map<String, Object> search(AnomalySearchFilter searchFilter, int limit, int offset) {
+ Predicate predicate = Predicate.EQ("child", false);
+ if (searchFilter.getStartTime() != null) {
+ predicate = Predicate.AND(predicate, Predicate.LT("startTime", searchFilter.getEndTime()));
+ }
+ if (searchFilter.getEndTime() != null) {
+ predicate = Predicate.AND(predicate, Predicate.GT("endTime", searchFilter.getStartTime()));
+ }
+ // search by detections or subscription groups
+ Set<Long> detectionConfigIds = new HashSet<>();
+ Set<Long> subscribedDetectionConfigIds = new HashSet<>();
+ if (!searchFilter.getDetectionNames().isEmpty()) {
+ detectionConfigIds =
+ this.detectionConfigDAO.findByPredicate(Predicate.IN("name", searchFilter.getDetectionNames().toArray()))
+ .stream()
+ .map(DetectionConfigBean::getId)
+ .collect(Collectors.toSet());
+ }
+ if (!searchFilter.getSubscriptionGroups().isEmpty()) {
+ subscribedDetectionConfigIds = this.detectionAlertConfigDAO.findByPredicate(
+ Predicate.IN("name", searchFilter.getSubscriptionGroups().toArray()))
+ .stream()
+ .map(detectionAlertConfigDTO -> detectionAlertConfigDTO.getVectorClocks().keySet())
+ .flatMap(Collection::stream)
+ .collect(Collectors.toSet());
+ }
+ if (!searchFilter.getDetectionNames().isEmpty() && !searchFilter.getSubscriptionGroups().isEmpty()) {
+ // intersect the detection config ids if searching by both
+ detectionConfigIds.retainAll(subscribedDetectionConfigIds);
+ } else {
+ detectionConfigIds.addAll(subscribedDetectionConfigIds);
+ }
+ if (!searchFilter.getDetectionNames().isEmpty() || !searchFilter.getSubscriptionGroups().isEmpty()) {
+ // add the predicate using detection config id
+ if (detectionConfigIds.isEmpty()) {
+ // if detection not found, return empty result
+ return ImmutableMap.of("count", 0, "limit", limit, "offset", offset, "elements", Collections.emptyList());
+ }
+ predicate = Predicate.AND(predicate, Predicate.IN("detectionConfigId", detectionConfigIds.toArray()));
+ }
+
+ // search by datasets
+ if (!searchFilter.getDatasets().isEmpty()) {
+ List<Predicate> datasetPredicates = new ArrayList<>();
+ for (String dataset : searchFilter.getDatasets()) {
+ datasetPredicates.add(Predicate.LIKE("collection", "%" + dataset + "%"));
+ }
+ predicate = Predicate.AND(predicate, Predicate.OR(datasetPredicates.toArray(new Predicate[0])));
+ }
+ // search by metrics
+ if (!searchFilter.getMetrics().isEmpty()) {
+ predicate = Predicate.AND(predicate, Predicate.IN("metric", searchFilter.getMetrics().toArray()));
+ }
+ // search by ids
+ if (!searchFilter.getAnomalyIds().isEmpty()) {
+ predicate = Predicate.AND(predicate, Predicate.IN("baseId", searchFilter.getAnomalyIds().toArray()));
+ }
+
+ long count;
+ List<MergedAnomalyResultDTO> results;
+ if (searchFilter.getFeedbacks().isEmpty()) {
+ List<Long> anomalyIds = this.anomalyDAO.findIdsByPredicate(predicate)
+ .stream()
+ .sorted(Comparator.reverseOrder())
+ .collect(Collectors.toList());
+ count = anomalyIds.size();
+ results = anomalyIds.isEmpty() ? Collections.emptyList()
+ : this.anomalyDAO.findByIds(paginateResults(anomalyIds, offset, limit));
+ } else {
+ // filter by feedback types if requested
+ List<MergedAnomalyResultDTO> anomalies = this.anomalyDAO.findByPredicate(predicate);
+ Set<AnomalyFeedbackType> feedbackFilters =
+ searchFilter.getFeedbacks().stream().map(AnomalyFeedbackType::valueOf).collect(Collectors.toSet());
+ results = anomalies.stream()
+ .filter(anomaly -> (anomaly.getFeedback() == null && feedbackFilters.contains(NO_FEEDBACK)) || (
+ anomaly.getFeedback() != null && feedbackFilters.contains(anomaly.getFeedback().getFeedbackType())))
+ .sorted(Comparator.comparingLong(AbstractDTO::getId).reversed())
+ .collect(Collectors.toList());
+ count = results.size();
+ results = paginateResults(results, offset, limit);
+ }
+ return ImmutableMap.of("count", count, "limit", limit, "offset", offset, "elements", results);
+ }
+
+ private <T> List<T> paginateResults(List<T> list, int offset, int limit) {
Review comment:
+1, good to see you Bryan!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] jihaozh commented on a change in pull request #5778: [TE] The endpoint for searching anomalies and pagination
Posted by GitBox <gi...@apache.org>.
jihaozh commented on a change in pull request #5778:
URL: https://github.com/apache/incubator-pinot/pull/5778#discussion_r465255934
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/dashboard/resources/v2/anomalies/AnomalySearcher.java
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.dashboard.resources.v2.anomalies;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.pinot.thirdeye.constant.AnomalyFeedbackType;
+import org.apache.pinot.thirdeye.datalayer.bao.DetectionAlertConfigManager;
+import org.apache.pinot.thirdeye.datalayer.bao.DetectionConfigManager;
+import org.apache.pinot.thirdeye.datalayer.bao.MergedAnomalyResultManager;
+import org.apache.pinot.thirdeye.datalayer.dto.AbstractDTO;
+import org.apache.pinot.thirdeye.datalayer.dto.MergedAnomalyResultDTO;
+import org.apache.pinot.thirdeye.datalayer.pojo.DetectionConfigBean;
+import org.apache.pinot.thirdeye.datalayer.util.Predicate;
+import org.apache.pinot.thirdeye.datasource.DAORegistry;
+
+import static org.apache.pinot.thirdeye.constant.AnomalyFeedbackType.*;
+
+
+/**
+ * The type Anomaly searcher.
+ */
+public class AnomalySearcher {
+ private final MergedAnomalyResultManager anomalyDAO;
+ private final DetectionConfigManager detectionConfigDAO;
+ private final DetectionAlertConfigManager detectionAlertConfigDAO;
+
+ /**
+ * Instantiates a new Anomaly searcher.
+ */
+ public AnomalySearcher() {
+ this.anomalyDAO = DAORegistry.getInstance().getMergedAnomalyResultDAO();
+ this.detectionConfigDAO = DAORegistry.getInstance().getDetectionConfigManager();
+ this.detectionAlertConfigDAO = DAORegistry.getInstance().getDetectionAlertConfigManager();
+ }
+
+ /**
+ * Search and retrieve all the anomalies matching to the search filter and limits.
+ *
+ * @param searchFilter the search filter
+ * @param limit the limit
+ * @param offset the offset
+ * @return the result
+ */
+ public Map<String, Object> search(AnomalySearchFilter searchFilter, int limit, int offset) {
+ Predicate predicate = Predicate.EQ("child", false);
+ if (searchFilter.getStartTime() != null) {
+ predicate = Predicate.AND(predicate, Predicate.LT("startTime", searchFilter.getEndTime()));
+ }
+ if (searchFilter.getEndTime() != null) {
+ predicate = Predicate.AND(predicate, Predicate.GT("endTime", searchFilter.getStartTime()));
+ }
+ // search by detections or subscription groups
+ Set<Long> detectionConfigIds = new HashSet<>();
+ Set<Long> subscribedDetectionConfigIds = new HashSet<>();
+ if (!searchFilter.getDetectionNames().isEmpty()) {
+ detectionConfigIds =
+ this.detectionConfigDAO.findByPredicate(Predicate.IN("name", searchFilter.getDetectionNames().toArray()))
+ .stream()
+ .map(DetectionConfigBean::getId)
+ .collect(Collectors.toSet());
+ }
+ if (!searchFilter.getSubscriptionGroups().isEmpty()) {
+ subscribedDetectionConfigIds = this.detectionAlertConfigDAO.findByPredicate(
+ Predicate.IN("name", searchFilter.getSubscriptionGroups().toArray()))
+ .stream()
+ .map(detectionAlertConfigDTO -> detectionAlertConfigDTO.getVectorClocks().keySet())
+ .flatMap(Collection::stream)
+ .collect(Collectors.toSet());
+ }
+ if (!searchFilter.getDetectionNames().isEmpty() && !searchFilter.getSubscriptionGroups().isEmpty()) {
+ // intersect the detection config ids if searching by both
+ detectionConfigIds.retainAll(subscribedDetectionConfigIds);
+ } else {
+ detectionConfigIds.addAll(subscribedDetectionConfigIds);
+ }
+ if (!searchFilter.getDetectionNames().isEmpty() || !searchFilter.getSubscriptionGroups().isEmpty()) {
+ // add the predicate using detection config id
+ if (detectionConfigIds.isEmpty()) {
+ // if detection not found, return empty result
+ return ImmutableMap.of("count", 0, "limit", limit, "offset", offset, "elements", Collections.emptyList());
+ }
+ predicate = Predicate.AND(predicate, Predicate.IN("detectionConfigId", detectionConfigIds.toArray()));
+ }
+
+ // search by datasets
+ if (!searchFilter.getDatasets().isEmpty()) {
+ List<Predicate> datasetPredicates = new ArrayList<>();
+ for (String dataset : searchFilter.getDatasets()) {
+ datasetPredicates.add(Predicate.LIKE("collection", "%" + dataset + "%"));
+ }
+ predicate = Predicate.AND(predicate, Predicate.OR(datasetPredicates.toArray(new Predicate[0])));
+ }
+ // search by metrics
+ if (!searchFilter.getMetrics().isEmpty()) {
+ predicate = Predicate.AND(predicate, Predicate.IN("metric", searchFilter.getMetrics().toArray()));
+ }
+ // search by ids
+ if (!searchFilter.getAnomalyIds().isEmpty()) {
+ predicate = Predicate.AND(predicate, Predicate.IN("baseId", searchFilter.getAnomalyIds().toArray()));
+ }
+
+ long count;
+ List<MergedAnomalyResultDTO> results;
+ if (searchFilter.getFeedbacks().isEmpty()) {
+ List<Long> anomalyIds = this.anomalyDAO.findIdsByPredicate(predicate)
+ .stream()
+ .sorted(Comparator.reverseOrder())
+ .collect(Collectors.toList());
+ count = anomalyIds.size();
+ results = anomalyIds.isEmpty() ? Collections.emptyList()
+ : this.anomalyDAO.findByIds(paginateResults(anomalyIds, offset, limit));
+ } else {
+ // filter by feedback types if requested
+ List<MergedAnomalyResultDTO> anomalies = this.anomalyDAO.findByPredicate(predicate);
+ Set<AnomalyFeedbackType> feedbackFilters =
+ searchFilter.getFeedbacks().stream().map(AnomalyFeedbackType::valueOf).collect(Collectors.toSet());
+ results = anomalies.stream()
+ .filter(anomaly -> (anomaly.getFeedback() == null && feedbackFilters.contains(NO_FEEDBACK)) || (
+ anomaly.getFeedback() != null && feedbackFilters.contains(anomaly.getFeedback().getFeedbackType())))
+ .sorted(Comparator.comparingLong(AbstractDTO::getId).reversed())
+ .collect(Collectors.toList());
+ count = results.size();
+ results = paginateResults(results, offset, limit);
+ }
+ return ImmutableMap.of("count", count, "limit", limit, "offset", offset, "elements", results);
+ }
+
+ private <T> List<T> paginateResults(List<T> list, int offset, int limit) {
Review comment:
make sense
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org