Spring Data Elasticsearch聚合报错：sum类型聚合无法接收子聚合

技术栈：Spring Boot 3、Spring 6、Elastic Client 8.5.x，对接Elasticsearch服务版本为7.x

可正常执行的Elasticsearch DSL查询语句：

GET someIndex-*/_search
{
  "size": 0,
  "query": {
    "bool": {
      "filter": [
        {
          "range": {
            "@timestamp": {
              "gte": "1688688000000",
              "lte": "1688774399999",
              "format": "epoch_millis"
            }
          }
        },
        {
          "query_string": {
            "analyze_wildcard": true,
            "query": "name: alise"
          }
        }
      ]
    }
  },
  "aggs": {
    "by_term": {
      "terms": {
        "field": "Product_type.keyword",
        "size": 10,
        "order": {
          "total_sum_metric": "desc"
        },
        "min_doc_count": 1
      },
      "aggs": {
        "total_sum_metric": {
          "sum": {
            "field": "metric"
          }
        },
        "group_by_date": {
          "date_histogram": {
            "interval": "30s",
            "field": "@timestamp",
            "min_doc_count": 1,
            "extended_bounds": {
              "min": "1688688000000",
              "max": "1688774399999"
            },
            "format": "epoch_millis"
          },
          "aggs": {
            "total_sum_metric": {
              "sum": {
                "field": "metric"
              }
            }
          }
        }
      }
    }
  }
}

转换后的Spring Data Native Query代码：

public void retrieveApplicationDetails(String applicationName, @NonNull LocalDateTime startDateTime, @NonNull LocalDateTime endDateTime) {
    final var startmilliSecond = startDateTime
          .atZone(ZoneOffset.UTC)
          .toInstant()
          .toEpochMilli();
    final var endmilliSecond = endDateTime
          .minusMinutes(1)
          .atZone(ZoneOffset.UTC)
          .toInstant()
          .toEpochMilli();
    final var query = buildSearchQuery(applicationName, startmilliSecond, endmilliSecond);
    SearchHits<Object> search = elasticsearchOperations.search(query, Object.class, IndexCoordinates.of("Index-*"));
}

private static NativeQuery buildSearchQuery(String applicationName, long startmilliSecond, long endmilliSecond) {
    var rangeQuery = new Query.Builder()
          .range(ran -> ran
                .field("@timestamp")
                .format("epoch_millis")
                .gte(JsonData.of(startmilliSecond))
                .lte(JsonData.of(endmilliSecond)))
          .build();

    var queryString = new QueryStringQuery.Builder()
          .query("name: %s".formatted(applicationName))
          .analyzeWildcard(true)
          .build()
          ._toQuery();

    final var termsAggregation = TermsAggregation.of(termBuilder -> termBuilder
          .field("Product_type.keyword")
          .size(10)
          .minDocCount(1)
          .order(new NamedValue<>("total_sum_metric", SortOrder.Desc)));

    final var sumAggregation = SumAggregation.of(metricAggregation -> metricAggregation.field("metric"));

    final var query = QueryBuilders.bool(boolBuilder -> boolBuilder.filter(List.of(rangeQuery, queryString)));

    final var groupByDateAggregation = Aggregation.of(byDateBuilder -> byDateBuilder
          .dateHistogram(dateHistoBuilder -> dateHistoBuilder
                .field("@timestamp")
                .fixedInterval(intervalBuilder -> intervalBuilder.time("30s"))
                .minDocCount(1)
                .extendedBounds(extendsBoundBuilder -> extendsBoundBuilder
                      .min(FieldDateMath.of(minBuilder -> minBuilder.value((double) startmilliSecond)))
                      .max(FieldDateMath.of(maxBuilder -> maxBuilder.value((double) endmilliSecond))))
                .format("epoch_millis")

          )
          .aggregations("total_sum_metric", sumAggregation));

    final var byTermAggregation = Aggregation.of(aggBuilder -> aggBuilder
          .terms(termsAggregation)
          .aggregations("total_sum_metric", Aggregation.of(sumMetricBuilder -> sumMetricBuilder
                .sum(sumAggregation)
                .aggregations("group_by_date", groupByDateAggregation))));

    return NativeQuery
          .builder()
          .withQuery(query)
          .withAggregation("by_term", byTermAggregation)
          .withMaxResults(0)
          .withPageable(PageRequest.of(0, MAX_PAGE_REQUEST_SIZE))
          .build();
}

执行查询时出现错误：
Aggregator [total_sum_metric] of type [sum] cannot accept sub-aggregation

问题原因与修复 #

错误原因 #

你在构建byTermAggregation时，错误地将group_by_date聚合挂载到了total_sum_metric这个sum聚合下面。sum属于度量聚合，不支持包含子聚合；按照原始DSL的结构，total_sum_metric和group_by_date应该是同级的，都作为by_term这个terms聚合的子聚合。

修正后的代码 #

private static NativeQuery buildSearchQuery(String applicationName, long startmilliSecond, long endmilliSecond) {
    var rangeQuery = new Query.Builder()
          .range(ran -> ran
                .field("@timestamp")
                .format("epoch_millis")
                .gte(JsonData.of(startmilliSecond))
                .lte(JsonData.of(endmilliSecond)))
          .build();

    var queryString = new QueryStringQuery.Builder()
          .query("name: %s".formatted(applicationName))
          .analyzeWildcard(true)
          .build()
          ._toQuery();

    final var termsAggregation = TermsAggregation.of(termBuilder -> termBuilder
          .field("Product_type.keyword")
          .size(10)
          .minDocCount(1)
          .order(new NamedValue<>("total_sum_metric", SortOrder.Desc)));

    final var sumAggregation = SumAggregation.of(metricAggregation -> metricAggregation.field("metric"));

    final var query = QueryBuilders.bool(boolBuilder -> boolBuilder.filter(List.of(rangeQuery, queryString)));

    // 构建日期直方图聚合，内部包含sum子聚合
    final var groupByDateAggregation = Aggregation.of(byDateBuilder -> byDateBuilder
          .dateHistogram(dateHistoBuilder -> dateHistoBuilder
                .field("@timestamp")
                .fixedInterval(intervalBuilder -> intervalBuilder.time("30s"))
                .minDocCount(1)
                .extendedBounds(extendsBoundBuilder -> extendsBoundBuilder
                      .min(FieldDateMath.of(minBuilder -> minBuilder.value((double) startmilliSecond)))
                      .max(FieldDateMath.of(maxBuilder -> maxBuilder.value((double) endmilliSecond))))
                .format("epoch_millis")
          )
          .aggregations("total_sum_metric", sumAggregation));

    // 正确构建terms聚合：sum聚合和date聚合同级，都作为terms的子聚合
    final var byTermAggregation = Aggregation.of(aggBuilder -> aggBuilder
          .terms(termsAggregation)
          // 添加sum聚合作为terms的子聚合
          .aggregations("total_sum_metric", Aggregation.of(sumMetricBuilder -> sumMetricBuilder.sum(sumAggregation)))
          // 添加date直方图聚合，和sum聚合同级
          .aggregations("group_by_date", groupByDateAggregation));

    return NativeQuery
          .builder()
          .withQuery(query)
          .withAggregation("by_term", byTermAggregation)
          .withMaxResults(0)
          .withPageable(PageRequest.of(0, MAX_PAGE_REQUEST_SIZE))
          .build();
}

内容的提问来源于stack exchange，提问作者Waley