add support for multiple bucket trees (for windowing) and add aggregations to the corresponding bucket when required

Author: fupelaqu

bridge/src/main/scala/app/softnetwork/elastic/sql/bridge/ElasticAggregation.scala

@@ -151,7 +151,7 @@ package object bridge {
   ): Seq[AbstractAggregation] = {
     val notNestedAggregations = aggregations.filterNot(_.nested)
 
-    val notNestedBuckets = request.buckets.filterNot(_.nested)
+    val notNestedBuckets = request.bucketTree.filterNot(_.bucket.nested)
 
     val rootAggregations = notNestedAggregations match {
       case Nil =>
@@ -164,8 +164,8 @@ package object bridge {
           None,
           aggregations
         ) match {
-          case Some(b) => Seq(b)
-          case _       => Seq.empty
+          case Nil  => Seq.empty
+          case aggs => aggs
         }
         buckets
       case aggs =>
@@ -179,14 +179,18 @@ package object bridge {
         val buckets = ElasticAggregation.buildBuckets(
           notNestedBuckets,
           request.sorts -- directions.keys,
-          aggregations,
+          aggs,
           directions,
           request.having.flatMap(_.criteria),
           None,
           aggs
         ) match {
-          case Some(b) => Seq(b)
-          case _       => aggregations
+          case Nil => aggs.map(_.agg)
+          case aggs =>
+            if (request.groupBy.isEmpty && request.windowFunctions.exists(_.isWindowing))
+              notNestedAggregations.filter(_.bucketPath.isEmpty).map(_.agg) ++ aggs
+            else
+              aggs
         }
         buckets
     }
@@ -210,12 +214,14 @@ package object bridge {
 
     // Group nested buckets by their nested path
     val nestedGroupedBuckets =
-      request.buckets
-        .filter(_.nested)
-        .groupBy(
-          _.nestedBucket.getOrElse(
-            throw new IllegalArgumentException(
-              "Nested bucket must have a nested element"
+      request.bucketTree
+        .filter(_.bucket.nested)
+        .map(tree =>
+          tree.groupBy(
+            _.bucket.nestedBucket.getOrElse(
+              throw new IllegalArgumentException(
+                "Nested bucket must have a nested element"
+              )
             )
           )
         )
@@ -235,17 +241,16 @@ package object bridge {
 
           // Get the buckets for this nested element
           val nestedBuckets =
-            nestedGroupedBuckets.getOrElse(n.innerHitsName, Seq.empty)
+            nestedGroupedBuckets.map(_.getOrElse(n.innerHitsName, Seq.empty))
 
           val notRelatedAggregationsToBuckets = elasticAggregations
             .filterNot { ea =>
-              nestedBuckets.exists(nb => nb.identifier.path == ea.sourceField)
+              nestedBuckets.flatten.exists(nb => nb.bucket.identifier.path == ea.sourceField)
             }
-            .map(_.agg)
 
           val relatedAggregationsToBuckets = elasticAggregations
             .filter { ea =>
-              nestedBuckets.exists(nb => nb.identifier.path == ea.sourceField)
+              nestedBuckets.flatten.exists(nb => nb.bucket.identifier.path == ea.sourceField)
             }
             .map(_.agg)
 
@@ -273,8 +278,12 @@ package object bridge {
               Some(n),
               aggregations
             ) match {
-              case Some(b) => Seq(b)
-              case _       => notRelatedAggregationsToBuckets
+              case Nil => notRelatedAggregationsToBuckets.map(_.agg)
+              case aggs =>
+                if (request.groupBy.isEmpty && request.windowFunctions.exists(_.isWindowing))
+                  notRelatedAggregationsToBuckets.filter(_.bucketPath.isEmpty).map(_.agg) ++ aggs
+                else
+                  aggs
             }
 
           val children = n.children

bridge/src/main/scala/app/softnetwork/elastic/sql/bridge/package.scala

@@ -68,7 +68,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
           collection.immutable.Seq(single.sources: _*),
           sql = Some(sql.query)
         )
-        if (single.windowBuckets.nonEmpty)
+        if (single.windowFunctions.exists(_.isWindowing))
           searchWithWindowEnrichment(sql, single)
         else
           singleSearch(elasticQuery, single.fieldAliases, single.sqlAggregations)
@@ -131,7 +131,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
     val query = elasticQuery.query
     val indices = elasticQuery.indices.mkString(",")
 
-    logger.debug(
+    logger.info(
       s"🔍 Searching with query \n$elasticQuery\nin indices '$indices'"
     )
 
@@ -1168,7 +1168,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
     request
       .copy(
         select = request.select.copy(fields = request.windowFields),
-        groupBy = request.groupBy.map(_.copy(buckets = request.windowBuckets)),
+        groupBy = None, //request.groupBy.map(_.copy(buckets = request.windowBuckets)),
         orderBy = None, // Not needed for aggregations
         limit = None // Need all buckets
       )
@@ -1236,7 +1236,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
   ): SQLSearchRequest = {
 
     // Remove window function fields from SELECT
-    val baseFields = request.select.fields.filterNot(_.windows.nonEmpty)
+    val baseFields = request.select.fields.filterNot(_.identifier.hasWindow)
 
     // Create modified request
     val baseRequest = request

core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala

@@ -168,7 +168,8 @@ package object client extends SerializationApi {
     aggType: AggregationType.AggregationType,
     distinct: Boolean,
     sourceField: String,
-    windowing: Boolean
+    windowing: Boolean,
+    bucketPath: String
   ) {
     def multivalued: Boolean = aggType == AggregationType.ArrayAgg
     def singleValued: Boolean = !multivalued
@@ -191,7 +192,8 @@ package object client extends SerializationApi {
       aggType,
       agg.distinct,
       agg.sourceField,
-      agg.aggType.isWindowing
+      agg.aggType.isWindowing,
+      agg.bucketPath
     )
   }
 }

core/src/main/scala/app/softnetwork/elastic/client/package.scala

@@ -190,7 +190,8 @@ class ElasticConversionSpec extends AnyFlatSpec with Matchers with ElasticConver
           aggType = AggregationType.ArrayAgg,
           distinct = false,
           "name",
-          windowing = true
+          windowing = true,
+          ""
         )
       )
     ) match {
@@ -643,7 +644,8 @@ class ElasticConversionSpec extends AnyFlatSpec with Matchers with ElasticConver
           aggType = AggregationType.ArrayAgg,
           distinct = false,
           "name",
-          windowing = true
+          windowing = true,
+          ""
         )
       )
     ) match {