Removed the old aggregation parsing path (#1117)

* Removed the old aggregation parsing path in lieu of the new one * PR fixes
slackhq · Oct 16, 2024 · 0c61808 · 0c61808
1 parent 866d0e6
commit 0c61808
Show file tree

Hide file tree

Showing 62 changed files with 466 additions and 5,900 deletions.
diff --git a/astra/src/main/java/com/slack/astra/chunk/ReadOnlyChunkImpl.java b/astra/src/main/java/com/slack/astra/chunk/ReadOnlyChunkImpl.java
@@ -592,7 +592,6 @@ public SearchResult<T> query(SearchQuery query) {
       return logSearcher.search(
           query.dataset,
           query.howMany,
-          query.aggBuilder,
           query.queryBuilder,
           query.sourceFieldFilter,
           query.aggregatorFactoriesBuilder);

diff --git a/astra/src/main/java/com/slack/astra/chunk/ReadWriteChunk.java b/astra/src/main/java/com/slack/astra/chunk/ReadWriteChunk.java
@@ -282,7 +282,6 @@ public SearchResult<T> query(SearchQuery query) {
     return logSearcher.search(
         query.dataset,
         query.howMany,
-        query.aggBuilder,
         query.queryBuilder,
         query.sourceFieldFilter,
         query.aggregatorFactoriesBuilder);

diff --git a/astra/src/main/java/com/slack/astra/elasticsearchApi/OpenSearchRequest.java b/astra/src/main/java/com/slack/astra/elasticsearchApi/OpenSearchRequest.java
diff --git a/astra/src/main/java/com/slack/astra/logstore/opensearch/OpenSearchAdapter.java b/astra/src/main/java/com/slack/astra/logstore/opensearch/OpenSearchAdapter.java
diff --git a/astra/src/main/java/com/slack/astra/logstore/search/LogIndexSearcher.java b/astra/src/main/java/com/slack/astra/logstore/search/LogIndexSearcher.java
@@ -1,6 +1,5 @@
 package com.slack.astra.logstore.search;
 
-import com.slack.astra.logstore.search.aggregations.AggBuilder;
 import java.io.Closeable;
 import org.opensearch.index.query.QueryBuilder;
 import org.opensearch.search.aggregations.AggregatorFactories;
@@ -9,7 +8,6 @@ public interface LogIndexSearcher<T> extends Closeable {
   SearchResult<T> search(
       String dataset,
       int howMany,
-      AggBuilder aggBuilder,
       QueryBuilder queryBuilder,
       SourceFieldFilter sourceFieldFilter,
       AggregatorFactories.Builder aggregatorFactoriesBuilder);

diff --git a/astra/src/main/java/com/slack/astra/logstore/search/LogIndexSearcherImpl.java b/astra/src/main/java/com/slack/astra/logstore/search/LogIndexSearcherImpl.java
@@ -13,7 +13,6 @@
 import com.slack.astra.logstore.LogMessage.SystemField;
 import com.slack.astra.logstore.LogWireMessage;
 import com.slack.astra.logstore.opensearch.OpenSearchAdapter;
-import com.slack.astra.logstore.search.aggregations.AggBuilder;
 import com.slack.astra.metadata.schema.LuceneFieldDef;
 import com.slack.astra.util.JsonUtil;
 import java.io.IOException;
@@ -58,14 +57,6 @@ public class LogIndexSearcherImpl implements LogIndexSearcher<LogMessage> {
 
   private final ReferenceManager.RefreshListener refreshListener;
 
-  // This feature flag enables using OpenSearch to parse our aggregations, rather than using the
-  // home-rolled
-  // aggregation parsing we're currently using
-  private static final Boolean useOpenSearchAggregationParsing =
-      Boolean.parseBoolean(
-          System.getProperty("astra.query.useOpenSearchAggregationParsing", "false"));
-  ;
-
   @VisibleForTesting
   public static SearcherManager searcherManagerFromChunkId(String chunkId, BlobStore blobStore)
       throws IOException {
@@ -106,14 +97,15 @@ public void afterRefresh(boolean didRefresh) {
   public SearchResult<LogMessage> search(
       String dataset,
       int howMany,
-      AggBuilder aggBuilder,
       QueryBuilder queryBuilder,
       SourceFieldFilter sourceFieldFilter,
       AggregatorFactories.Builder aggregatorFactoriesBuilder) {
 
     ensureNonEmptyString(dataset, "dataset should be a non-empty string");
     ensureTrue(howMany >= 0, "hits requested should not be negative.");
-    ensureTrue(howMany > 0 || aggBuilder != null, "Hits or aggregation should be requested.");
+    ensureTrue(
+        howMany > 0 || aggregatorFactoriesBuilder != null,
+        "Hits or aggregation should be requested.");
 
     ScopedSpan span = Tracing.currentTracer().startScopedSpan("LogIndexSearcherImpl.search");
     span.tag("dataset", dataset);
@@ -132,20 +124,16 @@ public SearchResult<LogMessage> search(
 
         if (howMany > 0) {
           CollectorManager<TopFieldCollector, TopFieldDocs> topFieldCollector =
-              buildTopFieldCollector(howMany, aggBuilder != null ? Integer.MAX_VALUE : howMany);
+              buildTopFieldCollector(
+                  howMany, aggregatorFactoriesBuilder != null ? Integer.MAX_VALUE : howMany);
           MultiCollectorManager collectorManager;
 
-          if (aggregatorFactoriesBuilder != null && useOpenSearchAggregationParsing) {
+          if (aggregatorFactoriesBuilder != null) {
             collectorManager =
                 new MultiCollectorManager(
                     topFieldCollector,
                     openSearchAdapter.getCollectorManager(
                         aggregatorFactoriesBuilder, searcher, query));
-          } else if (aggBuilder != null) {
-            collectorManager =
-                new MultiCollectorManager(
-                    topFieldCollector,
-                    openSearchAdapter.getCollectorManager(aggBuilder, searcher, query));
           } else {
             collectorManager = new MultiCollectorManager(topFieldCollector);
           }
@@ -156,14 +144,16 @@ public SearchResult<LogMessage> search(
           for (ScoreDoc hit : hits) {
             results.add(buildLogMessage(searcher, hit, sourceFieldFilter));
           }
-          if (aggBuilder != null) {
+          if (aggregatorFactoriesBuilder != null) {
             internalAggregation = (InternalAggregation) collector[1];
           }
         } else {
           results = Collections.emptyList();
           internalAggregation =
               searcher.search(
-                  query, openSearchAdapter.getCollectorManager(aggBuilder, searcher, query));
+                  query,
+                  openSearchAdapter.getCollectorManager(
+                      aggregatorFactoriesBuilder, searcher, query));
         }
 
         elapsedTime.stop();

diff --git a/astra/src/main/java/com/slack/astra/logstore/search/SearchQuery.java b/astra/src/main/java/com/slack/astra/logstore/search/SearchQuery.java
@@ -1,6 +1,5 @@
 package com.slack.astra.logstore.search;
 
-import com.slack.astra.logstore.search.aggregations.AggBuilder;
 import java.util.List;
 import org.opensearch.index.query.QueryBuilder;
 import org.opensearch.search.aggregations.AggregatorFactories;
@@ -13,7 +12,6 @@ public class SearchQuery {
   public final AggregatorFactories.Builder aggregatorFactoriesBuilder;
   public final QueryBuilder queryBuilder;
   public final int howMany;
-  public final AggBuilder aggBuilder;
   public final List<String> chunkIds;
   public final SourceFieldFilter sourceFieldFilter;
   public final long startTimeEpochMs;
@@ -24,14 +22,12 @@ public SearchQuery(
       long startTimeEpochMs,
       long endTimeEpochMs,
       int howMany,
-      AggBuilder aggBuilder,
       List<String> chunkIds,
       QueryBuilder queryBuilder,
       SourceFieldFilter sourceFieldFilter,
       AggregatorFactories.Builder aggregatorFactoriesBuilder) {
     this.dataset = dataset;
     this.howMany = howMany;
-    this.aggBuilder = aggBuilder;
     this.chunkIds = chunkIds;
     this.queryBuilder = queryBuilder;
     this.sourceFieldFilter = sourceFieldFilter;
@@ -50,8 +46,6 @@ public String toString() {
         + howMany
         + ", chunkIds="
         + chunkIds
-        + ", aggBuilder="
-        + aggBuilder
         + ", queryBuilder="
         + queryBuilder
         + ", sourceFieldFilter="

diff --git a/astra/src/main/java/com/slack/astra/logstore/search/SearchResultAggregatorImpl.java b/astra/src/main/java/com/slack/astra/logstore/search/SearchResultAggregatorImpl.java
@@ -4,7 +4,6 @@
 import brave.Tracing;
 import com.slack.astra.logstore.LogMessage;
 import com.slack.astra.logstore.opensearch.AstraBigArrays;
-import com.slack.astra.logstore.opensearch.OpenSearchAdapter;
 import com.slack.astra.logstore.opensearch.ScriptServiceProvider;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -24,14 +23,6 @@ public class SearchResultAggregatorImpl<T extends LogMessage> implements SearchR
 
   private final SearchQuery searchQuery;
 
-  // This feature flag enables using OpenSearch to parse our aggregations, rather than using the
-  // home-rolled
-  // aggregation parsing we're currently using
-  private static final Boolean useOpenSearchAggregationParsing =
-      Boolean.parseBoolean(
-          System.getProperty("astra.query.useOpenSearchAggregationParsing", "false"));
-  ;
-
   public SearchResultAggregatorImpl(SearchQuery searchQuery) {
     this.searchQuery = searchQuery;
   }
@@ -65,14 +56,10 @@ public SearchResult<T> aggregate(List<SearchResult<T>> searchResults, boolean fi
       // The last aggregation should be indicated using the final aggregation boolean. This performs
       // some final pass "destructive" actions, such as applying min doc count or extended bounds.
       if (finalAggregation) {
-        if (searchQuery.aggregatorFactoriesBuilder != null
-            && this.useOpenSearchAggregationParsing) {
+        if (searchQuery.aggregatorFactoriesBuilder != null) {
           Collection<AggregationBuilder> aggregationBuilders =
               searchQuery.aggregatorFactoriesBuilder.getAggregatorFactories();
           pipelineTree = aggregationBuilders.iterator().next().buildPipelineTree();
-        } else {
-          pipelineTree =
-              OpenSearchAdapter.getAggregationBuilder(searchQuery.aggBuilder).buildPipelineTree();
         }
 
         reduceContext =