feat(sql): ARRAY<X> round trip, derived and WITH columns typed, struct order (SCHEMA-6)

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

@@ -102,15 +102,17 @@ private[client] object AggregateReading {
           single.fieldAliases,
           dates,
           HitReading.Default,
-          single.sources.headOption.getOrElse("")
+          single.sources.headOption.getOrElse(""),
+          HitReading.remedyOf(single.schema)
         )
       val keys = new java.util.HashMap[String, KeyReader]()
       single.buckets.foreach { b =>
         val name = b.name
         if (name.nonEmpty && !keys.containsKey(name)) {
           val reader =
             if (instants.contains(name)) InstantKey
-            else keyReader(name, MultiSearch.bucketType(b), dates, readers)
+            // a GROUP BY level's key by its type, a window level's by its partition type (A-28)
+            else keyReader(name, MultiSearch.levelType(single, b), dates, readers)
           if (reader ne null) keys.put(name, reader)
         }
       }
@@ -133,7 +135,7 @@ private[client] object AggregateReading {
             case AggregationType.Count | AggregationType.Sum | AggregationType.Avg |
                 AggregationType.Min | AggregationType.Max | AggregationType.BucketScript =>
               owners.get(name).foreach { field =>
-                val reader = metricReader(MultiSearch.fieldType(field))
+                val reader = metricReader(MultiSearch.fieldType(field, single))
                 if (reader ne null) metrics.put(name, reader)
               }
             case AggregationType.FirstValue | AggregationType.LastValue |
@@ -500,18 +502,24 @@ private[client] object PartitionKeys {
   def of(request: SingleSearch, context: ConversionContext): PartitionKeys = {
     val partitions = request.windowFunctions.flatMap(wf => wf.partitionBy.zip(wf.buckets))
     val names = partitions.map(_._1.aliasOrName).distinct.toArray
+    // SCHEMA-6 (A-28) -- the base side of a selected ARRAY partition column: its value as read
+    val asRead: java.util.HashMap[String, String] = context match {
+      case s: SchemaColumns => s.reading.asReadKeys
+      case _                => null
+    }
     val instants: java.util.HashMap[String, String] = context match {
       case s: SchemaColumns => s.reading.instants
       case _                => null
     }
     // the base side of an instant column reads its doc value's key; the cache side, whose reading
     // holds no instant, reads the name, under which its key reader put the bucket's instant
     val rowKeys: Array[String] =
-      if ((instants eq null) || instants.isEmpty) names
+      if (((instants eq null) || instants.isEmpty) && ((asRead eq null) || asRead.isEmpty)) names
       else
         names.map { name =>
-          val key = instants.get(name)
-          if (key eq null) name else key
+          val key = if (instants eq null) null else instants.get(name)
+          val asReadKey = if ((key ne null) || (asRead eq null)) null else asRead.get(name)
+          if (key ne null) key else if (asReadKey ne null) asReadKey else name
         }
     val keyOf = new Array[Any => Any](names.length)
     if (context.isInstanceOf[SchemaColumns]) {
@@ -520,7 +528,7 @@ private[client] object PartitionKeys {
         val name = names(i)
         if (rowKeys(i) eq name)
           partitions.find(_._1.aliasOrName == name).foreach { case (_, bucket) =>
-            MultiSearch.bucketType(bucket).sqlType match {
+            MultiSearch.partitionType(bucket).sqlType match {
               case SQLTypes.Timestamp | SQLTypes.Time => keyOf(i) = toMillis
               case _                                  =>
             }
@@ -591,7 +599,7 @@ private[client] object PartitionKeys {
       !id.nested && id.functions.isEmpty && name.nonEmpty && name.indexOf('.') < 0 &&
       bucket.name == name
     plain && {
-      val t = MultiSearch.bucketType(bucket)
+      val t = MultiSearch.partitionType(bucket)
       t.sqlType match {
         case SQLTypes.Date | SQLTypes.Timestamp | SQLTypes.Time =>
           DateRoute.of(t.sqlType, t.format, dates).isInstanceOf[DateRoute.AsStored]
@@ -605,6 +613,12 @@ private[client] object PartitionKeys {
     */
   private[client] def instantKey(name: String): String = "\u0000instant:" + name
 
+  /** SCHEMA-6 (A-28) -- the base row key a selected `ARRAY` partition column's value is read under
+    * AS READ (0.24.0's reading, the one its window aggregation key has), beside the column's typed
+    * value: led by a NUL, so no column's name, and dropped by the final projection.
+    */
+  private[client] def asReadKey(name: String): String = "\u0000asread:" + name
+
   /** The instant of a base row's `epoch_millis` doc value (a one-element array of its text): the
     * epoch milliseconds' decimal text, the spelling of [[AggregateReading]] 's instant key -- any
     * sub-millisecond fraction (a `date_nanos` value) left out, as its aggregation buckets it. No

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

@@ -550,6 +550,32 @@ trait ElasticConversion {
       if (found) Some(current) else None
     }
 
+  /** SCHEMA-6 (A-26) -- the values the path `segments` holds in `map` where a list stands on it (a
+    * list of objects): every list met flattened, in order, as Elasticsearch indexes the path; one
+    * value is that value, several a `List`, none `None`. A read with no reader of its own (a
+    * ranking window's sort key), as read.
+    */
+  private[client] def collectedValue(
+    segments: Array[String],
+    map: Map[String, Any]
+  ): Option[Any] = {
+    val out = ListBuffer.empty[Any]
+    def walk(v: Any, i: Int): Unit = v match {
+      case null      => ()
+      case l: Seq[_] => l.foreach(walk(_, i))
+      case m: Map[_, _] if i < segments.length =>
+        m.asInstanceOf[Map[String, Any]].get(segments(i)).foreach(walk(_, i + 1))
+      case other if i == segments.length => out += other
+      case _                             => ()
+    }
+    walk(map, 0)
+    out.toList match {
+      case Nil        => None
+      case one :: Nil => Some(one)
+      case several    => Some(several)
+    }
+  }
+
   /** The value a row carries under a FLAT key, when it holds one (SCHEMA-12, A-3). A flat `null` is
     * no value: Elasticsearch indexes none for it (a document `{"obj.a": null, "obj": {"a": 4}}`
     * holds 4 in doc values, the `fields` API, SUM and COUNT), so the nested spelling is read
@@ -757,7 +783,11 @@ trait ElasticConversion {
       } catch {
         case cell: HitReading.MultiValuedCell =>
           throw new IllegalArgumentException(
-            HitReading.multiValuedMessage(cell, Option(hit.get("_id")).map(_.asText()))
+            HitReading.multiValuedMessage(
+              cell,
+              Option(hit.get("_id")).map(_.asText()),
+              Option(hit.get("_index")).map(_.asText())
+            )
           )
       }
     }
@@ -786,6 +816,9 @@ trait ElasticConversion {
   ): ListMap[String, Any] = {
     val entries = ListMap.newBuilder[String, Any]
     var folds: ListBuffer[(String, String, JsonNode)] = null
+    // SCHEMA-6 (A-26) -- the selected paths of a parent read whole that holds a list of objects
+    var collectedPaths: ListBuffer[(String, Any)] = null
+    val listPaths = reading.listPaths
     val it = node.properties().iterator()
     while (it.hasNext) {
       val e = it.next()
@@ -796,6 +829,22 @@ trait ElasticConversion {
           if ((ignored ne null) && ignored.contains(key)) reading.onIgnored(key) else null
         val value = if (onIgnored ne null) onIgnored(e.getValue) else reader(e.getValue)
         entries += fieldAliases.getOrElse(key, key) -> value
+        // SCHEMA-6 (A-28) -- a selected ARRAY partition column's value as read, under the row key
+        // the window join reads
+        if (!reading.asReadKeys.isEmpty) {
+          val asReadKey = reading.asReadKeys.get(key)
+          if (asReadKey ne null) entries += asReadKey -> jsonNodeToAny(e.getValue, fieldAliases)
+        }
+        if (!listPaths.isEmpty && e.getValue.isArray) {
+          val selected = listPaths.get(key)
+          if (selected ne null) {
+            if (collectedPaths eq null) collectedPaths = ListBuffer.empty
+            selected.foreach { path =>
+              val values = HitReading.collected(e.getValue, path.segments)
+              collectedPaths += path.key -> (if (values eq null) null else path.reader(values))
+            }
+          }
+        }
       } else {
         val parent = reading.foldsUnder(key)
         if (parent ne null) {
@@ -805,7 +854,15 @@ trait ElasticConversion {
           entries += fieldAliases.getOrElse(key, key) -> jsonNodeToAny(e.getValue, fieldAliases)
       }
     }
-    val row = entries.result()
+    val read = entries.result()
+    // a path collected under a list of objects answers under its own key, unless the hit holds a
+    // non-null flat value of it (A-3 precedence, `flatValue`)
+    val row =
+      if (collectedPaths eq null) read
+      else
+        collectedPaths.foldLeft(read) { case (acc, (key, value)) =>
+          if (ElasticConversion.flatValue(acc, key).isDefined) acc else acc + (key -> value)
+        }
     if (folds eq null) row
     else
       folds.foldLeft(row) { case (acc, (parent, key, value)) =>
@@ -2028,7 +2085,21 @@ trait ElasticConversion {
           val processedHits = hits.map { hit =>
             val source =
               if (byField eq null) extractSource(hit, fieldAliases)
-              else readElement(hit, fieldAliases, byField)
+              else
+                try readElement(hit, fieldAliases, byField)
+                catch {
+                  // SCHEMA-6 (D6) -- a several-valued scalar column of a top hit (FIRST_VALUE,
+                  // LAST_VALUE, ARRAY_AGG) fails as a row's does: the top hit's document and index,
+                  // and the ALTER that declares the column a list
+                  case cell: HitReading.MultiValuedCell =>
+                    throw new IllegalArgumentException(
+                      HitReading.multiValuedMessage(
+                        cell,
+                        Option(hit.get("_id")).map(_.asText()),
+                        Option(hit.get("_index")).map(_.asText())
+                      )
+                    )
+                }
             if (hasMultipleValues) {
               if (isRanking) {
                 val metadata = extractHitId(hit)

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

@@ -1116,8 +1116,23 @@ class TableExecutor(
                     return Left(error)
                   }
                   // we update the schema based on the DQL select clause
-                  fromSchema
-                    .mergeWithSearch(single)
+                  val merged = fromSchema.mergeWithSearch(single)
+                  // SCHEMA-6 (D2) -- a column the SELECT declares a list of lists (`ARRAY_AGG` of a
+                  // list column) is refused by name, before the target is created or replaced
+                  merged.columns.iterator
+                    .flatMap(app.softnetwork.elastic.sql.schema.nestedArrayRefusal(_))
+                    .toList
+                    .headOption
+                    .foreach { reason =>
+                      val error = ElasticError(
+                        message = reason,
+                        statusCode = Some(400),
+                        operation = Some("schema")
+                      )
+                      logger.error(s"❌ ${error.message}")
+                      return Left(error)
+                    }
+                  merged
                     .copy(
                       name = indexName, // set index name
                       settings = fromSchema.settings // remove settings that cannot be copied