add flag indicating whether or not an aggregate is a window function with partitioning

Author: fupelaqu

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

@@ -122,12 +122,7 @@ trait ScrollApi extends ElasticClientHelpers {
   )(implicit system: ActorSystem): Source[(Map[String, Any], ScrollMetrics), NotUsed] = {
     sql.request match {
       case Some(Left(single)) =>
-        if (
-          single.windowFunctions.nonEmpty && (single.fields.nonEmpty || single.windowFunctions
-            .flatMap(_.fields)
-            .distinct
-            .size > 1)
-        )
+        if (single.windowFunctions.nonEmpty)
           return scrollWithWindowEnrichment(sql, single, config)
 
         val sqlRequest = single.copy(score = sql.score)

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

@@ -68,12 +68,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
           collection.immutable.Seq(single.sources: _*),
           sql = Some(sql.query)
         )
-        if (
-          single.windowFunctions.nonEmpty && (single.fields.nonEmpty || single.windowFunctions
-            .flatMap(_.fields)
-            .distinct
-            .size > 1)
-        )
+        if (single.windowBuckets.nonEmpty)
           searchWithWindowEnrichment(sql, single)
         else
           singleSearch(elasticQuery, single.fieldAliases, single.sqlAggregations)
@@ -829,7 +824,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
           s"✅ Successfully executed search with inner hits in indices '${elasticQuery.indices.mkString(",")}'"
         )
         ElasticResult.attempt {
-          new JsonParser().parse(response).getAsJsonObject
+          JsonParser.parseString(response).getAsJsonObject
         } match {
           case ElasticFailure(error) =>
             logger.error(
@@ -918,7 +913,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
           s"✅ Successfully executed multi-search inner hits with ${elasticQueries.queries.size} queries"
         )
         ElasticResult.attempt {
-          new JsonParser().parse(response).getAsJsonObject
+          JsonParser.parseString(response).getAsJsonObject
         } match {
           case ElasticFailure(error) =>
             logger.error(
@@ -1287,51 +1282,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
     aggregations: Map[String, ClientAggregation]
   ): WindowValues = {
 
-    val values = aggregations
-      .filter(_._2.window)
-      .map { wf =>
-        val fieldName = wf._1
-
-        val aggType = wf._2.aggType
-
-        val sourceField = wf._2.sourceField
-
-        // Get value from row (already processed by ElasticConversion)
-        val value = row.get(fieldName).orElse {
-          logger.warn(s"⚠️ Window function '$fieldName' not found in aggregation result")
-          None
-        }
-
-        val validatedValue =
-          value match {
-            case Some(m: Map[String, Any]) =>
-              m.get(sourceField) match {
-                case Some(v) =>
-                  aggType match {
-                    case AggregationType.ArrayAgg =>
-                      v match {
-                        case l: List[_] =>
-                          Some(l)
-                        case other =>
-                          logger.warn(
-                            s"⚠️ Expected List for ARRAY_AGG '$fieldName', got ${other.getClass.getSimpleName}"
-                          )
-                          Some(List(other)) // Wrap into a List
-                      }
-                    case _ => Some(v)
-                  }
-                case None =>
-                  None
-              }
-            case other =>
-              other
-          }
-
-        fieldName -> validatedValue
-      }
-      .collect { case (name, Some(value)) =>
-        name -> value
-      }
+    val values = extractAggregationValues(row, aggregations)
 
     WindowValues(values)
   }

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

@@ -137,13 +137,17 @@ package object client extends SerializationApi {
     * @param distinct
     *   - when the aggregation is multivalued define if its values should be returned distinct or
     *     not
+    * @param sourceField
+    *   - the source field of the aggregation
+    * @param windowing
+    *   - whether the aggregation is a window function with partitioning
     */
   case class ClientAggregation(
     aggName: String,
     aggType: AggregationType.AggregationType,
     distinct: Boolean,
     sourceField: String,
-    window: Boolean
+    windowing: Boolean
   ) {
     def multivalued: Boolean = aggType == AggregationType.ArrayAgg
     def singleValued: Boolean = !multivalued
@@ -166,7 +170,7 @@ package object client extends SerializationApi {
       aggType,
       agg.distinct,
       agg.sourceField,
-      agg.aggType.isInstanceOf[WindowFunction]
+      agg.aggType.isWindowing
     )
   }
 }

core/src/test/scala/app/softnetwork/elastic/client/ElasticConversionSpec.scala

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

sql/src/main/scala/app/softnetwork/elastic/sql/function/aggregate/package.scala

@@ -29,6 +29,10 @@ package object aggregate {
     override def hasAggregation: Boolean = true
 
     def isBucketScript: Boolean = false
+
+    /** Indicates whether this aggregation is a windowing function with partitioning or not
+      */
+    def isWindowing: Boolean = false
   }
 
   case object COUNT extends Expr("COUNT") with AggregateFunction
@@ -97,6 +101,8 @@ package object aggregate {
     def window: Window
     def limit: Option[Limit]
 
+    override def isWindowing: Boolean = buckets.nonEmpty
+
     lazy val buckets: Seq[Bucket] = partitionBy.map(identifier => Bucket(identifier, None))
 
     lazy val bucketNames: Map[String, Bucket] = buckets.map { b =>
@@ -120,12 +126,16 @@ package object aggregate {
       val updated = this
         .withPartitionBy(partitionBy = partitionBy.map(_.update(request)))
       updated.withFields(
-        fields = request.select.fields
-          .filterNot(field =>
-            field.isAggregation || request.bucketNames.keys.toSeq
-              .contains(field.identifier.identifierName)
-          )
-          .filterNot(f => request.excludes.contains(f.sourceField))
+        fields = if (isWindowing) {
+          request.select.fields
+            .filterNot(field =>
+              field.isAggregation || request.bucketNames.keys.toSeq
+                .contains(field.identifier.identifierName)
+            )
+            .filterNot(f => request.excludes.contains(f.sourceField))
+        } else {
+          updated.fields
+        }
       )
     }
   }

sql/src/main/scala/app/softnetwork/elastic/sql/query/SQLSearchRequest.scala

@@ -122,7 +122,7 @@ case class SQLSearchRequest(
   }
 
   lazy val fields: Seq[String] = {
-    if (aggregates.isEmpty && buckets.isEmpty)
+    if (buckets.isEmpty)
       select.fields
         .filterNot(_.isScriptField)
         .filterNot(_.nested)
@@ -140,7 +140,7 @@ case class SQLSearchRequest(
   lazy val aggregates: Seq[Field] =
     select.fields
       .filter(f => f.isAggregation || f.isBucketScript)
-      .filterNot(_.windows.isDefined) ++ windowFields
+      .filterNot(_.isWindow) ++ windowFields
 
   lazy val sqlAggregations: Map[String, SQLAggregation] =
     aggregates.flatMap(f => SQLAggregation.fromField(f, this)).map(a => a.aggName -> a).toMap

sql/src/main/scala/app/softnetwork/elastic/sql/query/Select.scala

@@ -73,7 +73,7 @@ case class Field(
   lazy val windows: Option[WindowFunction] =
     functions.collectFirst { case th: WindowFunction => th }
 
-  def isWindow: Boolean = windows.isDefined
+  def isWindow: Boolean = windows.nonEmpty //.exists(_.partitionBy.nonEmpty)
 
   def update(request: SQLSearchRequest): Field = {
     windows match {