From e322c04e014d75ef7198b8e1373e3ca8bcab232e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Manciot?= Date: Thu, 8 Oct 2026 19:21:52 +0200 Subject: [PATCH] feat(sql): expose the set-operation reader to the relational engine; UNION ALL reads agreeing dates of a non-default format The relational engine (softclient4es-arrow) must answer a set operation with the same verdict, types and values as core. Core's typing and value reading were private, so arrow re-implemented them and drifted. Core now exposes them, private[elastic], so both use ONE implementation. - MultiSearch.columnTree(requests, operators): each step of the operator tree with its type per column, and an `untyped` flag where DuckDB 1.5.5 gives that step no type. columnTypes and branchTypes are unchanged. - SetOperationValues.legReaders(requests, operators, version, unreadable): per leg and column, a function that reads a value as its branch's declared type, as the cluster's Elasticsearch major indexed it, at every depth, and also where the branches agree. One class per type. A value with several elements fails, as at a converting position. `unreadable` (Fail by default, or KeepAsStored) decides what happens to a date the reader cannot read. - Core's UNION ALL reads an agreeing date position, at every depth, when a branch's field has a non-default mapping format (epoch_second, a custom pattern), so the column holds dates instead of the stored numbers and text. A value the reader cannot read is kept as stored, so nothing core answered before fails. Ordinary agreeing columns are untouched, at no per-row cost. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../elastic/client/SetOperationValues.scala | 524 +++++++++++++++--- .../client/SetOperationLegReadersSpec.scala | 438 +++++++++++++++ .../client/SetOperationMappedDatesSpec.scala | 281 ++++++++++ documentation/sql/dql_statements.md | 7 +- .../elastic/sql/query/package.scala | 246 ++++++-- .../query/SetOperationColumnTreeSpec.scala | 290 ++++++++++ .../sql/query/SetOperationTypeRuleSpec.scala | 80 +++ .../elastic/client/SetOperationTypeSpec.scala | 168 ++++++ 8 files changed, 1931 insertions(+), 103 deletions(-) create mode 100644 core/src/test/scala/app/softnetwork/elastic/client/SetOperationLegReadersSpec.scala create mode 100644 core/src/test/scala/app/softnetwork/elastic/client/SetOperationMappedDatesSpec.scala create mode 100644 sql/src/test/scala/app/softnetwork/elastic/sql/query/SetOperationColumnTreeSpec.scala diff --git a/core/src/main/scala/app/softnetwork/elastic/client/SetOperationValues.scala b/core/src/main/scala/app/softnetwork/elastic/client/SetOperationValues.scala index e98faf0d..af8cfb77 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/SetOperationValues.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/SetOperationValues.scala @@ -16,13 +16,14 @@ package app.softnetwork.elastic.client +import app.softnetwork.elastic.sql.operator.SetOperator import app.softnetwork.elastic.sql.query.MultiSearch import app.softnetwork.elastic.sql.query.MultiSearch.{ SetOperationColumn, SetOperationType, StructField } -import app.softnetwork.elastic.sql.query.SingleSearch +import app.softnetwork.elastic.sql.query.{SingleSearch, TemporalLiterals} import app.softnetwork.elastic.sql.`type`.{SQLType, SQLTypes} import java.nio.charset.StandardCharsets @@ -49,28 +50,41 @@ import scala.jdk.CollectionConverters._ * a mix, which is what JDBC and Arrow, typing a column from its first value, would otherwise * report wrongly. * + * The same converter, in its other mode, reads each leg of a set operation AS ITS BRANCH DECLARES + * IT, at every depth ([[legReaders]]): the reader the relational engine (arrow) types its legs + * with, and core's `UNION ALL` reads a date column of a mapping format of its own with. + * * 🔴 PER-ROW work on the extraction path, so it is decided ONCE per statement: [[converters]] is * `None` for every leg of a statement whose columns all agree -- no function, no rebuild -- and, - * for a leg that does convert, holds a function only at the positions whose column converts. The - * functions are applied as `rowProjector` rebuilds the row it builds anyway, never by a second - * pass. + * for a leg that does convert, holds a function only at the positions whose column converts or + * reads a mapping format. The functions are applied as `rowProjector` rebuilds the row it builds + * anyway, never by a second pass. */ -private[client] object SetOperationValues { +private[elastic] object SetOperationValues { /** The per-position conversions of each leg of `multiple`, in leg order -- `None` for a leg with * nothing to convert. `multiple` carries its branches' schemas (`resolveWithSchema`). * + * - A column whose branches disagree ([[SetOperationColumn.converts]]): every branch's value + * converted to the column's type ([[positionConverter]]). + * - A column whose branches agree, holding -- at the column or in a field or an element at any + * depth -- a date field with a mapping format of its own ([[readsMappedDates]]): the dates + * there read as their branches declare them ([[mappedDates]]) -- the date Elasticsearch + * indexed, never the JSON spelling -- every other value as read; a date core cannot read, + * and a multi-valued value, as stored. + * - Any other column: its values as read. + * * `version` is the Elasticsearch version the statement runs on (`VersionApi.version`): a date * field's values are read as THAT major reads its mapping format ([[MappedDateFormat]]). It is * asked at most ONCE per statement, and only when a converting position reads a date field. */ - def converters( + private[client] def converters( multiple: MultiSearch, version: => Option[String] ): Seq[Option[Array[Any => Any]]] = MultiSearch.columnTypes(multiple.requests) match { - case Right(columns) if columns.exists(_.converts) => - val names = columnNames(multiple, columns.size) + case Right(columns) if columns.exists(c => c.converts || readsMappedDates(c)) => + val names = columnNames(multiple.requests, columns.size) val dates = DateReading(version) multiple.requests.indices.map(leg => legConverters(multiple.requests(leg), leg, columns, names, dates) @@ -78,6 +92,126 @@ private[client] object SetOperationValues { case _ => multiple.requests.map(_ => None) } + /** THE READER of each leg of a set operation, column by column -- in leg order, `None` for a leg + * with nothing to read -- or, `Left`, the statement's refusal (`MultiSearch.branchTypes`' + * message word for word). + * + * THE CONTRACT read by the relational engine (arrow, `app.softnetwork.elastic.arrow`), which + * types each leg of a set operation by its branch's declared type, as core's `UNION ALL` reads + * it -- which is why it is `private[elastic]`: + * + * - INPUT: `MultiSearch.columnTypes`' -- the branches in text order, each with its schema + * attached, and `MultiSearch.resolvedOperators` (`Nil` is `UNION ALL` everywhere) -- and the + * Elasticsearch version the statement runs on (`VersionApi.version`, cached per client), + * asked at most ONCE, and only when a reader reads a date field. + * - Each function reads ONE raw value of its leg's column, as the response parser gives it + * (`ElasticConversion.jsonNodeToAny`), AS ITS BRANCH DECLARES IT + * (`SetOperationColumn.branches`), with its mapping format, as Elasticsearch indexed it -- + * at every depth, where the branches agree too. One class per type: an `INT` is an + * `Integer`, a `BIGINT` a `Long`, a `SMALLINT` a `Short`, a `TINYINT` a `Byte`, a `DOUBLE` a + * `Double`, a `REAL` a `Float`, a `BOOLEAN` a `Boolean`, a `DATE` a `LocalDate`, a + * `TIMESTAMP` a `ZonedDateTime` in UTC, a `TIME` a `LocalTime`, a `VARCHAR` a `String`, a + * `VARBINARY` the UTF-8 bytes of its base64 text, a `STRUCT` a `ListMap` of its fields in + * the branch's order, a `GEO_POINT` a `ListMap` of `lat` and `lon` (`Double`s), an `ARRAY` a + * `List` of its element's class. + * - Elasticsearch's one-element array (`script_fields`) is ONE value, its empty array NULL. + * - A value core cannot read fails the statement, in the words core's `UNION ALL` uses: a date + * its format does not read, or does not read without guessing (a zone name), a point that is + * not one, a multi-valued value at a scalar position -- where the branches agree too, in the + * words of a converting position ([[nonScalarFailure]]). + * - `unreadable` ([[Unreadable]]) says what a DATE core cannot read answers at a position + * whose branches AGREE: [[Unreadable.Fail]], the default, fails the statement as above; + * [[Unreadable.KeepAsStored]] answers the value exactly as the response parser gives it -- + * not of the position's class, which the caller must then type as it was stored -- as core's + * own `UNION ALL` answers it there. Where the branches disagree such a date always fails, as + * core's `UNION ALL` fails it. + * - `null` -- or `None` for the whole leg -- where nothing is read: a column whose type is not + * known (`ANY`), a branch that declares nothing (`SELECT *`), a `NULL` literal. + * + * Built ONCE per leg; no decision is made per row. + */ + private[elastic] def legReaders( + requests: Seq[SingleSearch], + operators: Seq[SetOperator], + version: => Option[String], + unreadable: Unreadable = Unreadable.Fail + ): Either[String, Seq[Option[Array[Any => Any]]]] = + MultiSearch.columnTypes(requests, operators).map { columns => + val names = columnNames(requests, columns.size) + val dates = DateReading(version) + requests.indices.map { leg => + val search = requests(leg) + // a leg's rows are built from its declared projection, position by position; a + // projection of another width -- an opaque `SELECT *` -- has no position to read at + if (search.select.fieldsWithComputedAliases.size != columns.size) None + else { + val readers: Array[Any => Any] = columns.indices.map { position => + legReader( + columns(position), + leg, + names(position), + fieldOf(search, position), + dates, + unreadable + ) + }.toArray + if (readers.forall(_ == null)) None else Some(readers) + } + } + } + + /** What a leg's reader ([[legReaders]]) answers for a DATE it cannot read -- one its mapping + * format does not read, or does not read without guessing (a zone name, a pattern letter no + * measurement covers), or any date of a format when the Elasticsearch version is not known -- at + * a position whose branches AGREE. Where the branches disagree, the value converts, and such a + * date FAILS the statement whatever this says, as core's `UNION ALL` fails it. + */ + private[elastic] sealed trait Unreadable + + private[elastic] object Unreadable { + + /** The statement fails, naming the value, the field and the format. */ + case object Fail extends Unreadable + + /** The value is answered AS STORED, exactly as the response parser gives it (a text, a number, + * Elasticsearch's array) -- as core's own `UNION ALL` answers it where the branches agree. + */ + case object KeepAsStored extends Unreadable + } + + /** How [[positionConverter]] treats a value. */ + private[client] sealed trait Reading + + private[client] object Reading { + + /** Converted to the position's type: core's `UNION ALL` where the branches disagree. */ + case object Converted extends Reading + + /** Read as the branch declares it, at every depth ([[legReaders]]); `unreadable` where the + * branches agree. + */ + final case class Declared(unreadable: Unreadable) extends Reading + } + + /** Whether core's `UNION ALL` reads the dates of a column whose branches AGREE ([[mappedDates]]): + * the column, or a field or an element at any depth inside it, is a date field with a mapping + * format of its own, not Elasticsearch's default (`epoch_second`, `yyyy/MM/dd`, ...), whose + * values the response parser gives as their JSON spells them (`1706697000`, `"2024/01/31"`), + * never as the date Elasticsearch indexed. Only a date field carries a format; a column of the + * default format keeps its values as read, at no cost; beside a `SELECT *` branch, whose values + * are not typed here, nothing is read. Decided once per statement. + */ + private def readsMappedDates(column: SetOperationColumn): Boolean = + !column.converts && column.branches.forall(_.isDefined) && + column.branches.exists(_.exists(mappedInside)) + + /** A date field's mapping format of its own, not Elasticsearch's default. */ + private def mapped(t: SetOperationType): Boolean = + t.format.exists(_ != TemporalLiterals.DefaultDateFormat) + + private def mappedInside(t: SetOperationType): Boolean = + mapped(t) || t.fields.exists(f => mappedInside(f.valueType)) || t.element.exists(mappedInside) + /** How a statement reads its date fields: as the Elasticsearch major it runs on reads their * mapping format ([[MappedDateFormat.Dialect]]). The version is asked once, the first time a * converting position reads a date field, and never for a statement that has none. @@ -99,12 +233,18 @@ private[client] object SetOperationValues { } /** The name each column takes: the first declaring branch's, as SQL and DuckDB name it. */ - private def columnNames(multiple: MultiSearch, width: Int): Seq[String] = - multiple.requests + private def columnNames(requests: Seq[SingleSearch], width: Int): Seq[String] = + requests .map(_.select.fieldsWithComputedAliases) .collectFirst { case fields if fields.size == width => fields.map(_.outputName) } .getOrElse(Seq.fill(width)("")) + /** The field a leg's item reads at `position` -- a failure names it -- or its output name. */ + private def fieldOf(leg: SingleSearch, position: Int): String = { + val item = leg.select.fieldsWithComputedAliases(position) + Option(item.identifier.name).filter(_.nonEmpty).getOrElse(item.outputName) + } + private def legConverters( leg: SingleSearch, index: Int, @@ -118,16 +258,198 @@ private[client] object SetOperationValues { else { val functions: Array[Any => Any] = columns.zipWithIndex.map { case (column, position) => // a NULL literal's branch too: its value is Elasticsearch's `[null]`, read as NULL - if (!column.converts || column.branches.lift(index).flatten.isEmpty) null - else { - val item = leg.select.fieldsWithComputedAliases(position) - val field = Option(item.identifier.name).filter(_.nonEmpty).getOrElse(item.outputName) - positionConverter(column.result, column.branches, index, names(position), field, dates) - } + if (column.branches.lift(index).flatten.isEmpty) null + else if (column.converts) + positionConverter( + column.result, + column.branches, + index, + names(position), + fieldOf(leg, position), + dates + ) + else if (readsMappedDates(column)) + mappedDates( + column.result, + column.branches, + index, + names(position), + fieldOf(leg, position), + dates + ).orNull + else null }.toArray if (functions.forall(_ == null)) None else Some(functions) } + /** Core's own `UNION ALL` at a position whose branches AGREE ([[readsMappedDates]]): the reading + * of the dates of a mapping format of their own at it, at every depth -- `None` where it holds + * none, its values then as read, at no cost. + * + * - A date position one of whose branches has a format of its own: each branch's date read as + * its branch declares it ([[legReaders]]' reader), the one of the default format too, so the + * position holds one class; a date core cannot read ([[Unreadable.KeepAsStored]]) and a + * multi-valued value ([[severalAsRead]]) as stored. + * - An object holding such a position: the same object, every other entry as read, in its + * order ([[objectPatch]]); a list of them, or of such dates: each element so + * ([[listPatch]]). + * - Any other position: as read. + * + * Decided ONCE per leg: only an object that holds such a date is rebuilt. + */ + private def mappedDates( + target: SetOperationType, + branches: Seq[Option[SetOperationType]], + index: Int, + name: String, + field: String, + dates: DateReading + ): Option[Any => Any] = + branches.lift(index).flatten.flatMap { own => + val path = if (field.isEmpty) name else field + own.sqlType match { + case SQLTypes.Struct => + val patches = own.fields.flatMap { f => + target.fields.find(_.name.equalsIgnoreCase(f.name)).flatMap { t => + mappedDates( + t.valueType, + SetOperationColumn.fieldTypes(branches, f.name), + index, + name, + s"$path.${f.name}", + dates + ).map(f.name -> _) + } + } + if (patches.isEmpty) None else Some(objectPatch(patches)) + case SQLTypes.Array(_) => + mappedDates( + target.element.getOrElse(SetOperationType(SQLTypes.Any)), + SetOperationColumn.elementTypes(branches), + index, + name, + path, + dates + ).map(listPatch) + case _ + if own.format.isDefined && target.sqlType.isTemporal && branches + .exists(_.exists(mapped)) => + Some( + severalAsRead( + positionConverter( + target, + branches, + index, + name, + path, + dates, + Reading.Declared(Unreadable.KeepAsStored), + agrees = Some(true) + ) + ) + ) + case _ => None + } + } + + /** An object whose fields `patches` name are read by their function, whatever their case -- every + * other entry as read, in its order. An object wrapped in Elasticsearch's array is read in it; + * an array of several objects -- a multi-valued field -- and any other value are answered as + * read. + */ + private def objectPatch(patches: Seq[(String, Any => Any)]): Any => Any = { + val names = patches.map(_._1).toArray + val reads = patches.map(_._2).toArray + def readOf(key: String): Any => Any = { + var i = 0 + while (i < names.length) { + if (names(i).equalsIgnoreCase(key)) return reads(i) + i += 1 + } + null + } + def patched(map: scala.collection.Map[_, _]): ListMap[String, Any] = { + val builder = ListMap.newBuilder[String, Any] + map.foreach { case (k, v) => + val key = String.valueOf(k) + val read = readOf(key) + builder += key -> (if (read eq null) v else read(v)) + } + builder.result() + } + def one(value: Any): Any = + value match { + case m: scala.collection.Map[_, _] => patched(m) + case m: java.util.Map[_, _] => patched(m.asScala) + case other => other + } + { + case s: Seq[_] if s.lengthCompare(1) == 0 => List(one(s.head)) + case l: java.util.List[_] if l.size() == 1 => List(one(l.get(0))) + case value => one(value) + } + } + + /** A list each element of which is read by `element`; a single value -- Elasticsearch stores a + * list of one as the value itself -- is read as it is. + */ + private def listPatch(element: Any => Any): Any => Any = { + case null => null + case s: Seq[_] => s.toList.map(element) + case l: java.util.List[_] => l.asScala.toList.map(element) + case one => element(one) + } + + /** A reader as core's own `UNION ALL` applies it where the branches agree ([[mappedDates]]): a + * value of several values -- a multi-valued field, an array of them in Elasticsearch's array + * (`[[1706697000, 1706697001]]`), an object -- is answered as stored there, as it always was, + * where the reader would fail it (the lead's ruling of 2026-10-08); every other value is read. + * The value is judged as the reader sees it: Elasticsearch's one-element arrays unwrapped. + */ + private def severalAsRead(read: Any => Any): Any => Any = { + @scala.annotation.tailrec + def several(value: Any): Boolean = + value match { + case s: Seq[_] => + if (s.lengthCompare(1) == 0) several(s.head) else s.lengthCompare(1) > 0 + case l: java.util.List[_] => + if (l.size() == 1) several(l.get(0)) else l.size() > 1 + case _: scala.collection.Map[_, _] | _: java.util.Map[_, _] => true + case _ => false + } + value => if (several(value)) value else read(value) + } + + /** The reader of branch `index`'s values at ONE column ([[legReaders]]): [[positionConverter]] + * reading each value as the branch declares it, at every depth -- `null` where nothing is read: + * a column whose type is not known (`ANY`), a branch that declares nothing there (`SELECT *`), a + * `NULL` literal. + */ + private def legReader( + column: SetOperationColumn, + index: Int, + name: String, + field: String, + dates: DateReading, + unreadable: Unreadable + ): Any => Any = + column.branches.lift(index).flatten match { + case Some(branch) if branch.sqlType != SQLTypes.Null && column.resultType != SQLTypes.Any => + positionConverter( + column.result, + column.branches, + index, + name, + field, + dates, + Reading.Declared(unreadable), + // the column's branches agree as core's `UNION ALL` reads them: a `SELECT *` branch + // beside them converts nothing + agrees = Some(!column.converts) + ) + case _ => null + } + /** The conversion of branch `index`'s value at ONE converting position -- a column, a field of a * struct, the element of a list -- to the position's type, by ONE converter at every depth: * EVERY branch's value at a converting position is converted, the branch whose type is already @@ -146,6 +468,12 @@ private[client] object SetOperationValues { * Inside a converting struct or list, a position whose branches all agree keeps its values as * read ([[SetOperationColumn.converts]]); a point is always read as a point. Decided ONCE per * leg, never per row. + * + * [[Reading.Declared]], the leg's READER ([[legReaders]]): the value is read AS ITS BRANCH + * DECLARES IT -- a scalar as the branch's type, a struct as the branch's fields in the branch's + * order, a point as a point -- at EVERY depth, where the branches agree too; `target`, the + * position's type, then names the position's type in a failure, and says whether the branches + * agree there -- `agrees`, where given, says it instead -- for a date core cannot read. */ private[client] def positionConverter( target: SetOperationType, @@ -153,29 +481,74 @@ private[client] object SetOperationValues { index: Int, name: String, field: String = "", - dates: DateReading = DateReading.unknown - ): Any => Any = - target.sqlType match { - case SQLTypes.GeoPoint => value => geoPoint(value, name) - case SQLTypes.Struct if target.fields.isEmpty => identity - case SQLTypes.Struct => structConverter(target, branches, index, name, field, dates) - case SQLTypes.Array(_) => listConverter(target, branches, index, name, field, dates) - case result => - val branch = branches.lift(index).flatten + dates: DateReading = DateReading.unknown, + reading: Reading = Reading.Converted, + agrees: Option[Boolean] = None + ): Any => Any = { + val branch = branches.lift(index).flatten + val read = reading != Reading.Converted + // the type the value is read as: the position's, or, for a leg's reader, the branch's own + val as = if (read) branch.getOrElse(target) else target + as.sqlType match { + case SQLTypes.GeoPoint => value => geoPoint(value, name) + // a STRUCT whose fields are not known: its values as read + case SQLTypes.Struct if as.fields.isEmpty || target.fields.isEmpty => identity + case SQLTypes.Struct => + structConverter(target, branches, index, name, field, dates, reading) + case SQLTypes.Array(_) => + listConverter(target, branches, index, name, field, dates, reading) + case scalar => val source = branch.fold[SQLType](SQLTypes.Null)(_.sqlType) - converter( + // a date core cannot read is answered as stored where the reader is told so and the + // branches agree -- never where they disagree (the lead's ruling of 2026-10-08) + val keep = reading match { + case Reading.Declared(Unreadable.KeepAsStored) => + agrees.getOrElse(!SetOperationColumn.converts(target, branches.flatten)) + case _ => false + } + // a value of several values fails, in the words of a converting position -- read as + // declared where the branches agree too: a leg typed by its branch's scalar type cannot + // carry a list, which core's own `UNION ALL` answers as read there (the lead's ruling of + // 2026-10-08) + val convert = converter( source, - result, - nonScalarFailure(result, branches.map(_.map(_.sqlType)), index, source, name), + scalar, + nonScalarFailure(target.sqlType, branches.map(_.map(_.sqlType)), index, source, name), branch.flatMap(_.format), if (field.isEmpty) name else field, name, - dates + dates, + keepUnreadable = keep ) + val canonical = + if (read && scalar == SQLTypes.Timestamp) convert.andThen(utc) else convert + if (keep) asStored(canonical) else canonical } + } + + /** A date core cannot read, answered by its reader as [[Unread]]: the value AS STORED, exactly as + * the reader received it -- Elasticsearch's array and all. + */ + private def asStored(read: Any => Any): Any => Any = + value => + read(value) match { + case Unread => value + case other => other + } + + /** What a date reader answers for a date it cannot read, where it keeps the value as stored. */ + private case object Unread + + /** A `TIMESTAMP` a leg's reader reads: in UTC, whatever offset its text carried. */ + private val utc: Any => Any = { + case z: ZonedDateTime if z.getZone != ZoneOffset.UTC => z.withZoneSameInstant(ZoneOffset.UTC) + case other => other + } /** The conversion at a position INSIDE a converting one: [[positionConverter]] where its branches - * disagree, and for a point; its values as read where they agree. + * disagree, and for a point; where they agree, its values as read, its dates of a mapping format + * of their own read ([[mappedDates]]) -- read as the branch declares them, with + * [[Reading.Declared]], wherever they agree or not. */ private def innerConverter( target: SetOperationType, @@ -183,13 +556,14 @@ private[client] object SetOperationValues { index: Int, name: String, field: String, - dates: DateReading + dates: DateReading, + reading: Reading ): Any => Any = if ( - target.sqlType == SQLTypes.GeoPoint || + reading != Reading.Converted || target.sqlType == SQLTypes.GeoPoint || SetOperationColumn.converts(target, branches.flatten) - ) positionConverter(target, branches, index, name, field, dates) - else identity + ) positionConverter(target, branches, index, name, field, dates, reading) + else mappedDates(target, branches, index, name, field, dates).getOrElse(identity) /** The failure of a converting position that a NON-SCALAR value -- a multi-valued field, an * object -- reaches: the position holds ONE value of its type per row, and such a value has no @@ -256,6 +630,9 @@ private[client] object SetOperationValues { * DuckDB's words, as a multi-valued value does in every converting position; so does a value * that is not an object. Decided ONCE per leg: the source field each field reads, and its * conversion. + * + * With [[Reading.Declared]] (a leg's reader, [[positionConverter]]), the map holds the BRANCH's + * fields, in its order and under its names, each read as the branch declares it. */ private[client] def structConverter( target: SetOperationType, @@ -263,29 +640,41 @@ private[client] object SetOperationValues { index: Int, name: String, field: String = "", - dates: DateReading = DateReading.unknown + dates: DateReading = DateReading.unknown, + reading: Reading = Reading.Converted ): Any => Any = { val source = branches.lift(index).flatten val sourceFields = source.fold[Seq[StructField]](Nil)(_.fields) - val read: Any => scala.collection.Map[String, Any] = + val mapOf: Any => scala.collection.Map[String, Any] = if (source.exists(_.sqlType == SQLTypes.GeoPoint)) value => geoPoint(value, name, target.fields) else value => objectOf(value, name, sourceFields, target.fields) val path = if (field.isEmpty) name else field - val plan: Seq[(String, Option[(String, Any => Any)])] = target.fields.map { t => - t.name -> sourceFields.find(_.name.equalsIgnoreCase(t.name)).map { f => - f.name -> innerConverter( - t.valueType, - SetOperationColumn.fieldTypes(branches, t.name), - index, - name, - s"$path.${f.name}", - dates - ) - } - } + def inner(t: SetOperationType, f: StructField): Any => Any = + innerConverter( + t, + SetOperationColumn.fieldTypes(branches, f.name), + index, + name, + s"$path.${f.name}", + dates, + reading + ) + val plan: Seq[(String, Option[(String, Any => Any)])] = + if (reading != Reading.Converted) + // the branch's fields, each typed at the position by the field of the column it is + sourceFields.map { f => + val t = target.fields.find(_.name.equalsIgnoreCase(f.name)).fold(f.valueType)(_.valueType) + f.name -> Some(f.name -> inner(t, f)) + } + else + target.fields.map { t => + t.name -> sourceFields.find(_.name.equalsIgnoreCase(t.name)).map { f => + f.name -> inner(t.valueType, f) + } + } value => - read(value) match { + mapOf(value) match { case null => null case map => ListMap(plan.map { case (field, from) => @@ -305,7 +694,8 @@ private[client] object SetOperationValues { index: Int, name: String, field: String = "", - dates: DateReading = DateReading.unknown + dates: DateReading = DateReading.unknown, + reading: Reading = Reading.Converted ): Any => Any = if (branches.lift(index).flatten.forall(_.sqlType == SQLTypes.Null)) _ => null else { @@ -315,7 +705,8 @@ private[client] object SetOperationValues { index, name, if (field.isEmpty) name else field, - dates + dates, + reading ) val list: Any => Any = { case null => null @@ -480,7 +871,7 @@ private[client] object SetOperationValues { * value to convert, and FAILS the statement ([[nonScalarFailure]]). A scalar of a class the * conversion does not expect is returned as it is. */ - def converter(source: SQLType, target: SQLType): Any => Any = + private[client] def converter(source: SQLType, target: SQLType): Any => Any = converter( source, target, @@ -494,6 +885,7 @@ private[client] object SetOperationValues { /** [[converter]] for a branch that reads a date FIELD, whose mapping `format` -- its own, or * Elasticsearch's default -- reads its values as the major `dates` names reads them * ([[readAs]]); `field` is its path and `column` the column it reaches, which a failure names. + * With `keepUnreadable`, a date the format does not read is [[Unread]] instead of a failure. */ private[client] def converter( source: SQLType, @@ -502,7 +894,8 @@ private[client] object SetOperationValues { format: Option[String], field: String, column: String, - dates: DateReading + dates: DateReading, + keepUnreadable: Boolean = false ): Any => Any = { val convert: Any => Any = target match { case SQLTypes.Varchar => text(source) @@ -516,7 +909,7 @@ private[client] object SetOperationValues { case SQLTypes.VarBinary => bytes(source) case _ => identity } - readAs(source, format, field, column, dates) match { + readAs(source, format, field, column, dates, keepUnreadable) match { case None => scalarOf(convert, nonScalar) case Some(read) => scalarOf(read.andThen(convert), nonScalar) } @@ -548,11 +941,12 @@ private[client] object SetOperationValues { format: Option[String], field: String, column: String, - dates: DateReading + dates: DateReading, + keepUnreadable: Boolean ): Option[Any => Any] = source match { case SQLTypes.Date | SQLTypes.Timestamp | SQLTypes.Time if format.isDefined => - Some(dateByFormat(source, format.get, field, column, dates)) + Some(dateByFormat(source, format.get, field, column, dates, keepUnreadable)) case SQLTypes.TinyInt | SQLTypes.SmallInt | SQLTypes.Int | SQLTypes.BigInt => Some(integral) case SQLTypes.Double => Some(double) case SQLTypes.Real => Some(real) @@ -567,14 +961,16 @@ private[client] object SetOperationValues { * `TIMESTAMP` is the instant that major indexed, a `DATE` its day and a `TIME` its time of day, * in UTC. A value the format does not read -- or one core cannot read without guessing, or any * value when the major is not known -- FAILS the statement, naming the value, the field and the - * format: never a wrong date, never a value passed on unread. + * format: never a wrong date, never a value passed on unread -- or, with `keepUnreadable`, is + * [[Unread]], which its reader answers as stored ([[asStored]]). */ private def dateByFormat( source: SQLType, spec: String, field: String, column: String, - dates: DateReading + dates: DateReading, + keepUnreadable: Boolean ): Any => Any = { val typeName = source match { case SQLTypes.Date => "DATE" @@ -586,11 +982,13 @@ private[client] object SetOperationValues { case SQLTypes.Time => _.atZone(ZoneOffset.UTC).toLocalTime case _ => _.atZone(ZoneOffset.UTC) } - def failure(value: Any, detail: String): Nothing = - throw new IllegalArgumentException( - s"Conversion Error: Could not read '$value' as a $typeName when casting from source " + - s"column $column: $detail" - ) + def failure(value: Any, detail: => String): Any = + if (keepUnreadable) Unread + else + throw new IllegalArgumentException( + s"Conversion Error: Could not read '$value' as a $typeName when casting from source " + + s"column $column: $detail" + ) dates.dialect match { case None => val detail = diff --git a/core/src/test/scala/app/softnetwork/elastic/client/SetOperationLegReadersSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/SetOperationLegReadersSpec.scala new file mode 100644 index 00000000..e6236558 --- /dev/null +++ b/core/src/test/scala/app/softnetwork/elastic/client/SetOperationLegReadersSpec.scala @@ -0,0 +1,438 @@ +/* + * Copyright 2025 SOFTNETWORK + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package app.softnetwork.elastic.client + +import app.softnetwork.elastic.sql.{StringValue, Value} +import app.softnetwork.elastic.sql.parser.Parser +import app.softnetwork.elastic.sql.query.{MultiSearch, SingleSearch} +import app.softnetwork.elastic.sql.operator.SetOperator +import app.softnetwork.elastic.sql.schema.{Column, Table} +import app.softnetwork.elastic.sql.serialization.JacksonConfig +import app.softnetwork.elastic.sql.`type`.SQLTypes +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.time.{LocalDate, LocalTime, ZoneOffset, ZonedDateTime} +import scala.collection.immutable.ListMap +import scala.util.Try + +/** Each leg of a set operation READ AS ITS BRANCH DECLARES IT ([[SetOperationValues.legReaders]]), + * the reader the relational engine types its legs with: each value as Elasticsearch indexed it, in + * ONE class per type, at every depth, where the branches agree too. Every value goes the way core + * receives it: its JSON, read by the response parser (`ElasticConversion.jsonNodeToAny`). The + * expectations are the contract's table, MEASURED on Elasticsearch 8.18.3. + */ +class SetOperationLegReadersSpec extends AnyFlatSpec with Matchers { + + private def format(f: String): ListMap[String, Value[_]] = + ListMap[String, Value[_]]("format" -> StringValue(f)) + + private val schema = Table( + "t", + columns = List( + Column("c_es", SQLTypes.Timestamp, options = format("epoch_second")), + Column("c_em", SQLTypes.Timestamp, options = format("epoch_millis")), + Column("c_c1", SQLTypes.Timestamp, options = format("yyyy/MM/dd")), + Column("c_c3", SQLTypes.Timestamp, options = format("dd/MM/yyyy HH:mm")), + Column("c_def", SQLTypes.Timestamp), + Column("c_dfmt", SQLTypes.Date, options = format("dd/MM/yyyy HH:mm")), + Column("c_des", SQLTypes.Date, options = format("epoch_second")), + Column("c_z", SQLTypes.Timestamp, options = format("yyyy-MM-dd'T'HH:mm:ss z")), + Column("ts", SQLTypes.Timestamp), + Column("d", SQLTypes.Date), + Column("t", SQLTypes.Time, options = format("HH:mm:ss")), + Column("n", SQLTypes.Int), + Column("bg", SQLTypes.BigInt), + Column("sm", SQLTypes.SmallInt), + Column("ti", SQLTypes.TinyInt), + Column("x", SQLTypes.Double), + Column("f", SQLTypes.Real), + Column("b", SQLTypes.Boolean), + Column("s", SQLTypes.Keyword), + Column("bin", SQLTypes.VarBinary), + Column("geo", SQLTypes.GeoPoint), + Column( + "oz", + SQLTypes.Struct, + multiFields = + List(Column("z", SQLTypes.Timestamp, options = format("yyyy-MM-dd'T'HH:mm:ss z"))) + ), + Column("arr", SQLTypes.Array(SQLTypes.Int)), + Column( + "obj", + SQLTypes.Struct, + multiFields = List( + Column("n", SQLTypes.Int), + Column("e", SQLTypes.Timestamp, options = format("epoch_second")), + Column("s", SQLTypes.Keyword), + Column("p", SQLTypes.Struct, multiFields = List(Column("c", SQLTypes.Double))) + ) + ), + Column( + "obj2", + SQLTypes.Struct, + multiFields = List(Column("z", SQLTypes.Keyword), Column("n", SQLTypes.BigInt)) + ), + Column( + "items", + SQLTypes.Array(SQLTypes.Struct), + multiFields = List(Column("q", SQLTypes.Int), Column("d", SQLTypes.Date)) + ) + ) + ) + + private val es8 = Some("8.18.3") + + private def resolved(sql: String): (Seq[SingleSearch], Seq[SetOperator]) = + Parser(sql) match { + case Right(m: MultiSearch) => (m.requests.map(_.update(Some(schema))), m.resolvedOperators) + case other => fail(s"[$sql] expected a set operation, got $other") + } + + private def readers( + sql: String, + version: => Option[String] = es8 + ): Seq[Option[Array[Any => Any]]] = { + val (requests, operators) = resolved(sql) + SetOperationValues.legReaders(requests, operators, version) match { + case Right(legs) => legs + case Left(refusal) => fail(s"[$sql] refused: $refusal") + } + } + + /** A document's value as core receives it: its JSON, read by the response parser. */ + private def received(json: String): Any = + ElasticConversion.jsonNodeToAny(JacksonConfig.objectMapper.readTree(json), ListMap.empty) + + /** `SELECT AS k FROM t UNION ALL SELECT AS k FROM t` */ + private def union(a: String, b: String, operator: String = "UNION ALL"): String = + s"SELECT $a AS k FROM t $operator SELECT $b AS k FROM t" + + /** The reader of leg `leg`, column 1, of `sql`. */ + private def reader(sql: String, leg: Int = 0, version: => Option[String] = es8): Any => Any = + readers(sql, version)(leg) match { + case Some(functions) if functions(0) ne null => functions(0) + case other => fail(s"[$sql] leg $leg reads nothing: $other") + } + + /** A value and its class -- bytes by their content. */ + private def shown(v: Any): (Any, Class[_]) = + v match { + case null => (null, classOf[Null]) + case b: Array[Byte] => (b.toSeq, classOf[Array[Byte]]) + case other => (other, other.getClass) + } + + private def utc(s: String): ZonedDateTime = ZonedDateTime.parse(s) + + private def point(lat: Double, lon: Double) = ListMap("lat" -> lat, "lon" -> lon) + + "A leg's reader" should "read every value of the contract's table as Elasticsearch indexed it" in { + // (field, the JSON written, the value read) -- each field beside itself, its branches agreeing + val cases: Seq[(String, String, Any)] = Seq( + ("c_es", "1706697000", utc("2024-01-31T10:30:00Z")), + ("c_es", "\"1706697000.5\"", utc("2024-01-31T10:30:00.500Z")), + ("c_c1", "\"2024/01/31\"", utc("2024-01-31T00:00:00Z")), + ("c_c3", "\"31/01/2024 10:30\"", utc("2024-01-31T10:30:00Z")), + ("c_def", "\"2024\"", utc("2024-01-01T00:00:00Z")), + ("c_def", "2024", utc("2024-01-01T00:00:00Z")), + ("c_dfmt", "\"31/01/2024 23:30\"", LocalDate.parse("2024-01-31")), + ("c_des", "1706697000", LocalDate.parse("2024-01-31")), + // a zone-less text is in UTC, whatever the JVM's zone + ("ts", "\"2024-01-31T10:30:00\"", utc("2024-01-31T10:30:00Z")), + ("d", "1706697000000", LocalDate.parse("2024-01-31")), + ("n", "\"5.5\"", Integer.valueOf(5)), + ("n", "\"-5.9\"", Integer.valueOf(-5)), + ("x", "\"1e3\"", java.lang.Double.valueOf(1000.0)), + ("x", "\" 2.5 \"", java.lang.Double.valueOf(2.5)), + ("b", "\"\"", java.lang.Boolean.FALSE), + ("geo", "\"48.85,2.35\"", point(48.85, 2.35)), + ("geo", "[2.35,48.85]", point(48.85, 2.35)), + ("geo", "\"POINT (2.35 48.85)\"", point(48.85, 2.35)), + ("geo", "\"u09tv\"", point(48.8232421875, 2.3291015625)) + ) + val wrong = cases.flatMap { case (field, json, want) => + val got = Try(reader(union(field, field))(received(json))) + if (got.map(shown).toOption.contains(shown(want))) None + else Some(s"$field <- $json read $got instead of ${shown(want)}") + } + wrong shouldBe empty + // a zone name: core's failure, naming the value, the field and the format + intercept[IllegalArgumentException]( + reader(union("c_z", "c_z"))(received("\"2024-01-31T10:30:00 CET\"")) + ).getMessage shouldBe + "Conversion Error: Could not read '2024-01-31T10:30:00 CET' as a TIMESTAMP when casting from " + + "source column k: core cannot read it with the date format 'yyyy-MM-dd'T'HH:mm:ss z' of " + + "field c_z as Elasticsearch 8 does: the zone 'CET' is a region id or a zone name, whose " + + "offset comes from the zone data of the JVM reading it, not Elasticsearch's: core reads " + + "only Z, an offset, UTC, GMT and UT" + } + + it should "read each type in its ONE class, beside itself or not" in { + val items = + Seq("n", "bg", "sm", "ti", "x", "f", "b", "d", "ts", "t", "s", "bin", "obj", "geo", "arr") + val row: Seq[String] = Seq( + "5", + "5", + "5", + "5", + "5", + "0.5", + "true", + "\"2024-01-31\"", + "\"2024-01-31T12:30:00+02:00\"", + "\"10:30:00\"", + "\"a\"", + "\"YWI=\"", + """{"s":"x","n":1,"e":1706697000,"p":{"c":2}}""", + """{"lat":48.85,"lon":2.35}""", + "[1,2]" + ) + val want: Seq[Any] = Seq( + Integer.valueOf(5), + java.lang.Long.valueOf(5L), + java.lang.Short.valueOf(5.toShort), + java.lang.Byte.valueOf(5.toByte), + java.lang.Double.valueOf(5.0), + java.lang.Float.valueOf(0.5f), + java.lang.Boolean.TRUE, + LocalDate.parse("2024-01-31"), + utc("2024-01-31T10:30:00Z"), + LocalTime.parse("10:30:00"), + "a", + "YWI=".getBytes(java.nio.charset.StandardCharsets.UTF_8), + // the branch's fields, in its order + ListMap( + "e" -> utc("2024-01-31T10:30:00Z"), + "n" -> Integer.valueOf(1), + "p" -> ListMap("c" -> java.lang.Double.valueOf(2.0)), + "s" -> "x" + ), + point(48.85, 2.35), + List(Integer.valueOf(1), Integer.valueOf(2)) + ) + val select = items.mkString("SELECT ", ", ", " FROM t") + val legs = readers(s"$select UNION ALL $select") + legs should have size 2 + val wrong = for { + leg <- legs + functions <- leg.toSeq + ((item, json), (read, value)) <- items.zip(row).zip(functions.toSeq.zip(want)) + got = shown(read(received(json))) + if got != shown(value) + } yield s"$item <- $json read $got instead of ${shown(value)}" + wrong shouldBe empty + // a struct's ListMap keeps its order; a TIMESTAMP is in UTC + legs.head.get + .apply(12)(received(row(12))) + .asInstanceOf[ListMap[String, Any]] + .keys + .toSeq shouldBe + Seq("e", "n", "p", "s") + legs.head.get + .apply(8)(received(row(8))) + .asInstanceOf[ZonedDateTime] + .getZone shouldBe ZoneOffset.UTC + // beside another type, a leg is read as ITS branch declares it: the INT leg an Integer, the + // DOUBLE leg a Double -- the relational engine casts them + shown(reader(union("n", "x"), 0)(received("\"5.5\""))) shouldBe shown(Integer.valueOf(5)) + shown(reader(union("n", "x"), 1)(received("5"))) shouldBe shown(java.lang.Double.valueOf(5.0)) + shown(reader(union("n", "s"), 0)(received("5"))) shouldBe shown(Integer.valueOf(5)) + shown(reader(union("d", "ts"), 0)(received("\"2024-01-31\""))) shouldBe + shown(LocalDate.parse("2024-01-31")) + // a point beside an object: a point, {lat, lon} + reader(union("geo", "obj2"), 0)(received("\"48.85,2.35\"")) shouldBe point(48.85, 2.35) + // Elasticsearch's one-element array (`script_fields`) is ONE value, its empty array NULL + shown(reader(union("n", "n"))(List(Integer.valueOf(3)))) shouldBe shown(Integer.valueOf(3)) + reader(union("n", "n"))(Nil) shouldBe (null: Any) + reader(union("n", "n"))(null) shouldBe (null: Any) + } + + it should "read a struct as its branch declares it, at every depth, where the branches agree" in { + // obj beside obj: every field read, `e` with its own format, `n` truncated as ES indexed it + reader(union("obj", "obj"))( + received("""{"n":"5.5","e":"1706697000","p":{"c":"1e3"}}""") + ) shouldBe + ListMap( + "e" -> utc("2024-01-31T10:30:00Z"), + "n" -> 5, + "p" -> ListMap("c" -> 1000.0), + "s" -> null + ) + // obj beside obj2, a merged struct: each leg its OWN fields, in its order, under its names + reader(union("obj", "obj2"), 1)(received("""{"z":"a","n":"7"}""")) shouldBe + ListMap("n" -> java.lang.Long.valueOf(7L), "z" -> "a") + // a list of objects: each element its fields + reader(union("items", "items"))( + received("""[{"q":"2.5","d":"2024-01-31"},{"q":3}]""") + ) shouldBe List( + ListMap("d" -> LocalDate.parse("2024-01-31"), "q" -> 2), + ListMap("d" -> null, "q" -> 3) + ) + // one object stored bare is a list of one + reader(union("items", "items"))(received("""{"q":4}""")) shouldBe + List(ListMap("d" -> null, "q" -> 4)) + } + + it should "read a date as the Elasticsearch major the statement runs on reads it" in { + val ts = union("c_es", "c_es") + // a fraction of a second: Elasticsearch 8 and 7 keep its milliseconds, 6 truncates it + reader(ts, version = Some("8.18.3"))(received("\"1706697000.5\"")) shouldBe + utc("2024-01-31T10:30:00.500Z") + reader(ts, version = Some("7.17.29"))(received("\"1706697000.5\"")) shouldBe + utc("2024-01-31T10:30:00.500Z") + reader(ts, version = Some("6.8.23"))(received("\"1706697000.5\"")) shouldBe + utc("2024-01-31T10:30:00Z") + // Elasticsearch 7 reads no negative epoch_millis, Elasticsearch 8 does + reader(union("c_em", "c_em"))(received("\"-2024\"")) shouldBe utc("1969-12-31T23:59:57.976Z") + intercept[IllegalArgumentException]( + reader(union("c_em", "c_em"), version = Some("7.17.29"))(received("\"-2024\"")) + ).getMessage shouldBe + "Conversion Error: Could not read '-2024' as a TIMESTAMP when casting from source column k: " + + "the date format 'epoch_millis' of field c_em does not read it as Elasticsearch 7 does" + // the version not known: no date is guessed + intercept[IllegalArgumentException]( + reader(ts, version = None)(received("1706697000")) + ).getMessage shouldBe + "Conversion Error: Could not read '1706697000' as a TIMESTAMP when casting from source column " + + "k: the date format 'epoch_second' of field c_es reads it as the Elasticsearch version does, " + + "which is not known" + // the version asked ONCE per statement, and only when a reader reads a date field + var asked = 0 + def version: Option[String] = { asked += 1; es8 } + readers(s"SELECT c_es, ts, n FROM t UNION ALL SELECT c_c1, d, x FROM t", version) + asked shouldBe 1 + readers(union("n", "x"), version) + asked shouldBe 1 + } + + it should "fail, in core's words, on a value core cannot read" in { + def failure(sql: String, json: String, leg: Int = 0): String = + intercept[IllegalArgumentException](reader(sql, leg)(received(json))).getMessage + failure(union("c_c1", "c_c1"), "\"31-01-2024\"") shouldBe + "Conversion Error: Could not read '31-01-2024' as a TIMESTAMP when casting from source column " + + "k: the date format 'yyyy/MM/dd' of field c_c1 does not read it as Elasticsearch 8 does" + failure(union("geo", "geo"), "\"nowhere!\"") shouldBe + "Conversion Error: Could not read 'nowhere!' as a GEO_POINT when casting from source column " + + "k: a point is an object {lat, lon}, a text 'lat,lon', an array [lon, lat], a geohash or a " + + "WKT POINT (lon lat)" + failure(union("obj", "obj"), "\"x\"") shouldBe + "Conversion Error: Could not read 'x' as a STRUCT when casting from source column k" + // a field inside an object, named by its path + failure(union("obj", "obj"), """{"e":"31-01-2024"}""") should include( + "the date format 'epoch_second' of field obj.e does not read it" + ) + } + + it should "fail a multi-valued value at a scalar position, where the branches agree too" in { + def failure(sql: String, json: String, leg: Int = 0): String = + intercept[IllegalArgumentException](reader(sql, leg)(received(json))).getMessage + // where the branches agree: core's words for a converting position (DuckDB's) + failure(union("s", "s"), """["a","b"]""") shouldBe + "Conversion Error: Type VARCHAR can't be cast to the destination type VARCHAR[] when casting " + + "from source column k" + failure(union("n", "n"), "[1,2]") shouldBe + "Conversion Error: Unimplemented type for cast (INTEGER -> INTEGER[]) when casting from " + + "source column k" + failure(union("c_es", "c_es"), "[1706697000,1706697001]") shouldBe + "Conversion Error: Unimplemented type for cast (TIMESTAMP -> TIMESTAMP[]) when casting from " + + "source column k" + // inside an object + failure(union("obj", "obj"), """{"n":[1,2]}""") shouldBe + "Conversion Error: Unimplemented type for cast (INTEGER -> INTEGER[]) when casting from " + + "source column k" + // where they disagree: the same words as core's UNION ALL + val (requests, _) = resolved(union("n", "x")) + val converting = SetOperationValues.converters(MultiSearch(requests), es8) + val core = intercept[IllegalArgumentException](converting.head.get.apply(0)(List(1, 2))) + failure(union("n", "x"), "[1,2]") shouldBe core.getMessage + core.getMessage shouldBe + "Conversion Error: Unimplemented type for cast (DOUBLE -> INTEGER[]) when casting from source " + + "column k" + // a list position carries several values + reader(union("arr", "arr"))(received("[1,2,3]")) shouldBe List(1, 2, 3) + } + + it should "answer a date it cannot read as stored where told so, where the branches agree" in { + def keeping(sql: String, leg: Int = 0, version: Option[String] = es8): Any => Any = { + val (requests, operators) = resolved(sql) + SetOperationValues.legReaders( + requests, + operators, + version, + SetOperationValues.Unreadable.KeepAsStored + ) match { + case Right(legs) => legs(leg).get.apply(0) + case Left(refusal) => fail(s"[$sql] refused: $refusal") + } + } + val cet = "2024-01-31T10:30:00 CET" + // a zone name: the value as stored, exactly -- Elasticsearch's array too + keeping(union("c_z", "c_z"))(received(s"\"$cet\"")) shouldBe cet + keeping(union("c_z", "c_z"))(received(s"[\"$cet\"]")) shouldBe List(cet) + // a date it reads is read + shown(keeping(union("c_z", "c_z"))(received("\"2024-01-31T10:30:00 UTC\""))) shouldBe + shown(utc("2024-01-31T10:30:00Z")) + // inside an object whose branches agree + keeping(union("oz", "oz"))(received(s"""{"z":"$cet"}""")) shouldBe ListMap("z" -> cet) + // the version not known: as stored + keeping(union("c_es", "c_es"), version = None)(received("1706697000")) shouldBe 1706697000 + // beside a SELECT * branch the column converts nothing, as core's UNION ALL reads it + keeping(s"${union("c_z", "c_z")} UNION ALL SELECT * FROM t")(received(s"\"$cet\"")) shouldBe + cet + // where the branches disagree (a TIMESTAMP beside a DATE) it fails, as core's UNION ALL does + intercept[IllegalArgumentException]( + keeping(union("c_z", "d"))(received(s"\"$cet\"")) + ).getMessage should include("the zone 'CET' is a region id or a zone name") + // other failures are not dates: they fail + intercept[IllegalArgumentException]( + keeping(union("geo", "geo"))(received("\"nowhere!\"")) + ).getMessage should startWith("Conversion Error: Could not read 'nowhere!' as a GEO_POINT") + intercept[IllegalArgumentException](keeping(union("n", "n"))(received("[1,2]"))) + // the default fails + intercept[IllegalArgumentException](reader(union("c_z", "c_z"))(received(s"\"$cet\""))) + } + + it should "read nothing where nothing is typed: ANY, SELECT *, a NULL literal" in { + // a NULL literal: its slot is empty + readers(union("n", "NULL")).map(_.map(_.toSeq.map(_ ne null))) shouldBe + Seq(Some(Seq(true)), None) + // SELECT *: the leg reads nothing + readers("SELECT n FROM t UNION ALL SELECT * FROM t").map(_.isDefined) shouldBe Seq(true, false) + // a column whose type is not known: nothing read, in any leg + val (typed, _) = resolved(union("n", "n")) + val unknown = Parser(union("n", "n")) match { + case Right(m: MultiSearch) => m.requests + case other => fail(s"$other") + } + SetOperationValues.legReaders(Seq(typed.head, unknown(1)), Nil, es8) shouldBe + Right(Seq(None, None)) + // a refused statement: its refusal, word for word + val (refused, operators) = resolved(union("d", "n")) + SetOperationValues.legReaders(refused, operators, es8) shouldBe + MultiSearch.branchTypes(refused, operators).map(_ => Nil) + } + + it should "read every leg of INTERSECT and EXCEPT as declared, whatever the operator" in { + Seq("UNION", "INTERSECT", "EXCEPT ALL").foreach { operator => + shown(reader(union("c_es", "ts", operator))(received("1706697000"))) shouldBe + shown(utc("2024-01-31T10:30:00Z")) + shown(reader(union("n", "x", operator))(received("\"5.5\""))) shouldBe + shown(Integer.valueOf(5)) + } + } +} diff --git a/core/src/test/scala/app/softnetwork/elastic/client/SetOperationMappedDatesSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/SetOperationMappedDatesSpec.scala new file mode 100644 index 00000000..3e68ac39 --- /dev/null +++ b/core/src/test/scala/app/softnetwork/elastic/client/SetOperationMappedDatesSpec.scala @@ -0,0 +1,281 @@ +/* + * Copyright 2025 SOFTNETWORK + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package app.softnetwork.elastic.client + +import app.softnetwork.elastic.sql.{StringValue, Value} +import app.softnetwork.elastic.sql.parser.Parser +import app.softnetwork.elastic.sql.query.MultiSearch +import app.softnetwork.elastic.sql.schema.{Column, Table} +import app.softnetwork.elastic.sql.serialization.JacksonConfig +import app.softnetwork.elastic.sql.`type`.SQLTypes +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.time.{LocalDate, ZonedDateTime} +import scala.collection.immutable.ListMap + +/** Core's own `UNION ALL` reads a date whose branches AGREE as the date Elasticsearch indexed when + * one of its branches is a field with a mapping format of its own -- `epoch_second`, `yyyy/MM/dd`, + * ... -- never as its JSON spells it (`1706697000`, `"2024/01/31"`), at the column and at every + * depth inside it. A date of the default format, a value with no date, a multi-valued value and a + * date core cannot read keep their values as stored: no statement core answered fails. + */ +class SetOperationMappedDatesSpec extends AnyFlatSpec with Matchers { + + private def format(f: String): ListMap[String, Value[_]] = + ListMap[String, Value[_]]("format" -> StringValue(f)) + + private val schema = Table( + "t", + columns = List( + Column("c_es", SQLTypes.Timestamp, options = format("epoch_second")), + Column("c_c1", SQLTypes.Timestamp, options = format("yyyy/MM/dd")), + Column("c_def", SQLTypes.Timestamp), + Column("c_dfmt", SQLTypes.Date, options = format("dd/MM/yyyy HH:mm")), + Column("ts", SQLTypes.Timestamp), + Column("d", SQLTypes.Date), + Column("n", SQLTypes.Int), + Column("s", SQLTypes.Keyword), + Column("c_z", SQLTypes.Timestamp, options = format("yyyy-MM-dd VV")), + Column( + "obj", + SQLTypes.Struct, + multiFields = List( + Column("a", SQLTypes.Timestamp, options = format("epoch_second")), + Column("n", SQLTypes.Int), + Column("z", SQLTypes.Timestamp, options = format("yyyy-MM-dd VV")) + ) + ), + // `obj`'s fields, every date of the default format + Column( + "objd", + SQLTypes.Struct, + multiFields = List( + Column("a", SQLTypes.Timestamp), + Column("n", SQLTypes.Int), + Column("z", SQLTypes.Timestamp) + ) + ), + // a converting object beside `obj`: `n` an INT beside a BIGINT, `a` agreeing + Column( + "objl", + SQLTypes.Struct, + multiFields = List( + Column("a", SQLTypes.Timestamp, options = format("epoch_second")), + Column("n", SQLTypes.BigInt) + ) + ), + Column( + "deep", + SQLTypes.Struct, + multiFields = List( + Column( + "p", + SQLTypes.Struct, + multiFields = List(Column("c", SQLTypes.Date, options = format("yyyy/MM/dd"))) + ), + Column("q", SQLTypes.Keyword) + ) + ), + Column( + "items", + SQLTypes.Array(SQLTypes.Struct), + multiFields = List( + Column("d", SQLTypes.Date, options = format("dd/MM/yyyy")), + Column("q", SQLTypes.Int) + ) + ), + Column( + "itemsd", + SQLTypes.Array(SQLTypes.Struct), + multiFields = List(Column("d", SQLTypes.Date), Column("q", SQLTypes.Int)) + ), + Column("stamps", SQLTypes.Array(SQLTypes.Timestamp), options = format("epoch_second")) + ) + ) + + private def resolved(sql: String): MultiSearch = + Parser(sql) match { + case Right(m: MultiSearch) => m.copy(requests = m.requests.map(_.update(Some(schema)))) + case other => fail(s"[$sql] expected a set operation, got $other") + } + + private def plan(sql: String, version: Option[String] = Some("8.18.3")) = + SetOperationValues.converters(resolved(sql), version) + + private def union(a: String, b: String): String = + s"SELECT $a AS k FROM t UNION ALL SELECT $b AS k FROM t" + + /** A document's value as core receives it: its JSON, read by the response parser. */ + private def received(json: String): Any = + ElasticConversion.jsonNodeToAny(JacksonConfig.objectMapper.readTree(json), ListMap.empty) + + /** What leg `leg` of `sql` answers at column 1 for a value written as `json`. */ + private def answered( + sql: String, + leg: Int, + json: String, + version: Option[String] = Some("8.18.3") + ): Any = { + val value = received(json) + plan(sql, version)(leg) match { + case Some(functions) if functions(0) ne null => functions(0)(value) + case _ => value + } + } + + /** A value and its class. */ + private def shown(v: Any): (Any, Class[_]) = + if (v == null) (null, classOf[Null]) else (v, v.getClass) + + private def utc(s: String): ZonedDateTime = ZonedDateTime.parse(s) + + "An agreeing date column of a mapping format of its own" should "answer the date Elasticsearch indexed, in every leg" in { + // epoch_second beside a default-format TIMESTAMP: seconds, never 1970's milliseconds + shown(answered(union("c_es", "ts"), 0, "1706697000")) shouldBe + shown(utc("2024-01-31T10:30:00Z")) + shown(answered(union("c_es", "ts"), 0, "\"1706697000.5\"")) shouldBe + shown(utc("2024-01-31T10:30:00.500Z")) + // the default-format leg too: every row a timestamp + shown(answered(union("c_es", "ts"), 1, "\"2024-02-01T00:00:00Z\"")) shouldBe + shown(utc("2024-02-01T00:00:00Z")) + shown(answered(union("c_es", "ts"), 1, "\"2024\"")) shouldBe shown(utc("2024-01-01T00:00:00Z")) + // a custom pattern + shown(answered(union("c_c1", "ts"), 0, "\"2024/01/31\"")) shouldBe + shown(utc("2024-01-31T00:00:00Z")) + // a DATE field: its day + shown(answered(union("c_dfmt", "d"), 0, "\"31/01/2024 23:30\"")) shouldBe + shown(LocalDate.parse("2024-01-31")) + // as the major the statement runs on reads it: Elasticsearch 6 truncates the fraction + shown(answered(union("c_es", "ts"), 0, "\"1706697000.5\"", Some("6.8.23"))) shouldBe + shown(utc("2024-01-31T10:30:00Z")) + // a computed branch's one-element array is one value + shown(answered(union("c_es", "ts"), 1, "[\"2024-02-01T00:00:00Z\"]")) shouldBe + shown(utc("2024-02-01T00:00:00Z")) + } + + it should "keep a multi-valued value as read, as it always was" in { + answered(union("c_es", "ts"), 0, "[1706697000,1706697001]") shouldBe List( + 1706697000, + 1706697001 + ) + // in Elasticsearch's array (an array of arrays, which Elasticsearch accepts and flattens), or + // an object: as stored, judged as the reader sees it -- never a failure + answered(union("c_es", "ts"), 0, "[[1706697000,1706697001]]") shouldBe + List(List(1706697000, 1706697001)) + answered(union("c_c1", "ts"), 0, """[["2024/01/31","2024/02/01"]]""") shouldBe + List(List("2024/01/31", "2024/02/01")) + answered(union("c_es", "ts"), 0, """[{"x":1}]""") shouldBe List(ListMap("x" -> 1)) + answered(union("c_es", "ts"), 0, """[[{"x":1}]]""") shouldBe List(List(ListMap("x" -> 1))) + // one value in arrays of one: read + shown(answered(union("c_es", "ts"), 0, "[[1706697000]]")) shouldBe + shown(utc("2024-01-31T10:30:00Z")) + // inside an object + answered(union("obj", "obj"), 0, """{"a":[[1706697000,1706697001]]}""") shouldBe + ListMap("a" -> List(List(1706697000, 1706697001))) + } + + it should "change nothing else: the default format, a column with no date, a NULL, SELECT *" in { + // the control: two fields of the default format keep their values as read + plan(union("c_def", "ts")) shouldBe Seq(None, None) + answered(union("c_def", "ts"), 0, "\"2024\"") shouldBe "2024" + answered(union("c_def", "ts"), 0, "2024") shouldBe 2024 + // in one statement, only the date column of a format of its own is read + plan("SELECT c_es, n, s FROM t UNION ALL SELECT ts, n, s FROM t") + .map(_.map(_.toSeq.map(_ ne null))) shouldBe + Seq(Some(Seq(true, false, false)), Some(Seq(true, false, false))) + // a NULL literal reads nothing + plan(union("c_es", "NULL")).map(_.isDefined) shouldBe Seq(true, false) + // beside a SELECT * branch, whose values are not typed, nothing is read + plan("SELECT c_es AS k FROM t UNION ALL SELECT * FROM t") shouldBe Seq(None, None) + // objects and lists whose date fields all have the default format: as read, at no cost + plan(union("objd", "objd")) shouldBe Seq(None, None) + plan(union("itemsd", "itemsd")) shouldBe Seq(None, None) + } + + it should "keep a date it cannot read as stored, where a converting column fails it" in { + // a region id: its offset comes from zone data -- the value as stored, never a failure + answered(union("c_z", "ts"), 0, "\"2024-01-31 Europe/Paris\"") shouldBe + "2024-01-31 Europe/Paris" + answered(union("c_z", "ts"), 0, "[\"2024-01-31 Europe/Paris\"]") shouldBe + List("2024-01-31 Europe/Paris") + // the readable values of the column are still read + shown(answered(union("c_z", "ts"), 0, "\"2024-01-31 Z\"")) shouldBe + shown(utc("2024-01-31T00:00:00Z")) + shown(answered(union("c_z", "ts"), 1, "\"2024\"")) shouldBe + shown(utc("2024-01-01T00:00:00Z")) + // the Elasticsearch version not known: every date of a format as stored + answered(union("c_es", "ts"), 0, "1706697000", version = None) shouldBe 1706697000 + // a column that converts (a TIMESTAMP beside a DATE) fails it, as before + intercept[IllegalArgumentException]( + answered(union("c_z", "d"), 0, "\"2024-01-31 Europe/Paris\"") + ).getMessage should include( + "Could not read '2024-01-31 Europe/Paris' as a TIMESTAMP when casting from source column k: " + + "core cannot read it with the date format 'yyyy-MM-dd VV' of field c_z as Elasticsearch 8 does" + ) + } + + "An agreeing date field of a mapping format of its own" should "be read at every depth, every other entry as stored" in { + // an object's field: read; every other entry -- one the schema does not hold too -- as stored, + // in its order + answered(union("obj", "obj"), 0, """{"x":"kept","n":"5.5","a":1706697000}""") shouldBe + ListMap("x" -> "kept", "n" -> "5.5", "a" -> utc("2024-01-31T10:30:00Z")) + // the default-format branch's field too + answered(union("obj", "objd"), 1, """{"a":"2024","n":1}""") shouldBe + ListMap("a" -> utc("2024-01-01T00:00:00Z"), "n" -> 1) + // a date it cannot read inside: as stored + answered(union("obj", "obj"), 0, """{"z":"2024-01-31 Europe/Paris","a":1706697000}""") shouldBe + ListMap("z" -> "2024-01-31 Europe/Paris", "a" -> utc("2024-01-31T10:30:00Z")) + // an object in an object + answered(union("deep", "deep"), 0, """{"q":"x","p":{"c":"2024/01/31"}}""") shouldBe + ListMap("q" -> "x", "p" -> ListMap("c" -> LocalDate.parse("2024-01-31"))) + // a list of objects, and one stored bare + answered(union("items", "itemsd"), 0, """[{"d":"31/01/2024","q":"2"},{"q":3}]""") shouldBe + List(ListMap("d" -> LocalDate.parse("2024-01-31"), "q" -> "2"), ListMap("q" -> 3)) + answered(union("items", "itemsd"), 1, """{"d":"2024-01-31"}""") shouldBe + ListMap("d" -> LocalDate.parse("2024-01-31")) + // a list of dates + answered(union("stamps", "stamps"), 0, "[1706697000,\"1706697000.5\"]") shouldBe + List(utc("2024-01-31T10:30:00Z"), utc("2024-01-31T10:30:00.500Z")) + // an object wrapped in Elasticsearch's array is read in it; several objects as stored + answered(union("obj", "obj"), 0, """[{"a":1706697000}]""") shouldBe + List(ListMap("a" -> utc("2024-01-31T10:30:00Z"))) + answered(union("obj", "obj"), 0, """[{"a":1706697000},{"a":1}]""") shouldBe + List(ListMap("a" -> 1706697000), ListMap("a" -> 1)) + // NULL, and a value that is no object, as stored + answered(union("obj", "obj"), 0, "null") shouldBe (null: Any) + answered(union("obj", "obj"), 0, "\"x\"") shouldBe "x" + } + + it should "be read inside a converting object too, where its branches agree" in { + // `n` converts (an INT beside a BIGINT): the object is DuckDB's; `a` agrees and is read + answered(union("obj", "objl"), 0, """{"a":1706697000,"n":1}""") shouldBe + ListMap("a" -> utc("2024-01-31T10:30:00Z"), "n" -> java.lang.Long.valueOf(1L), "z" -> null) + } + + it should "ask the Elasticsearch version once, and only for a column it reads" in { + var asked = 0 + def version: Option[String] = { asked += 1; Some("8.18.3") } + SetOperationValues.converters(resolved(union("c_def", "ts")), version) + asked shouldBe 0 + SetOperationValues.converters( + resolved("SELECT c_es, c_c1 FROM t UNION ALL SELECT ts, ts FROM t"), + version + ) + asked shouldBe 1 + } +} diff --git a/documentation/sql/dql_statements.md b/documentation/sql/dql_statements.md index 1364953a..20a507cc 100644 --- a/documentation/sql/dql_statements.md +++ b/documentation/sql/dql_statements.md @@ -724,7 +724,12 @@ UTC. So `1706697000` in an `epoch_second` field is `2024-01-31 10:30:00`, not a second on Elasticsearch 6; `"2024/01/31"` in a `yyyy/MM/dd` field is a date, not a text; a format's `||` alternatives are tried in order, and the first one that parses a value decides; in the default format `"2024"` and `2024` are the year 2024. A value computed by the statement (a function, a -`CAST`) has no format: there, digits are epoch milliseconds and an ISO text is its instant. +`CAST`) has no format: there, digits are epoch milliseconds and an ISO text is its instant. A date +whose branches already agree -- a column, or a field of an object or of a list of objects at any +depth -- is read the same way, in every branch, as soon as one of its branches is a field with a +`format` of its own (`epoch_second`, `yyyy/MM/dd`, ...), every other value of the row as read, while +a date position whose branches all have the default format, a multi-valued value and a date core +cannot read (a zone name, say) are returned as stored. A value is read only where its reading is the same on every Java version core runs on (8, 11, 17 and 21) and equal to the instant the Elasticsearch major indexed, measured value by value on diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/query/package.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/query/package.scala index e7a9de0c..87dc88dc 100644 --- a/sql/src/main/scala/app/softnetwork/elastic/sql/query/package.scala +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/query/package.scala @@ -2823,6 +2823,36 @@ package object query { branches.map(_.flatMap(_.element)) } + /** One NODE of a set operation as DuckDB types it ([[columnTree]]): a run of one union kind, or + * the two sides of an `INTERSECT [ALL]` or an `EXCEPT [ALL]`, with the type it gives each + * column. The contract read by the relational engine (arrow), which types each node of its own + * statement the same way: + * + * @param operator + * the node's operator: the run's union kind (`UNION` is `UNION ALL`, `UNION_DISTINCT` + * `UNION`), or `INTERSECT [ALL]` / `EXCEPT [ALL]` + * @param operands + * the node's operands in text order: `Left(i)` the branch of index `i`, `Right(node)` a + * subtree -- a run holds every one of its operands, an `INTERSECT` or an `EXCEPT` its two + * sides + * @param columns + * the type the node gives each column, at every depth and with no date format, as + * [[SetOperationColumn.result]] holds it: a `STRUCT`'s merged fields in DuckDB's order; + * `ANY` where it is not known, and for a column no branch under the node declares + * @param untyped + * per column: whether DuckDB 1.5.5 gives the node NO type there, at the column or at any + * field or element inside it -- a `VARBINARY` beside a number, a temporal or a `BOOLEAN`, + * with a text or not, a `TIME` beside a date or a number -- which the rules answer as + * `VARCHAR` ([[typeNode]]): the relational engine then casts each operand of the node to + * [[columns]] itself, since DuckDB would fail on it + */ + final case class SetOperationNode( + operator: SetOperator, + operands: Seq[Either[Int, SetOperationNode]], + columns: Seq[SetOperationType], + untyped: Seq[Boolean] + ) + /** DuckDB's numeric lattice over core's types: a `BOOLEAN` reads as 1 / 0 beside any number, * and the wider type wins (`INTEGER` with `BIGINT` is `BIGINT`, with `REAL` is `REAL`, with * `DOUBLE` is `DOUBLE`). MEASURED on DuckDB 1.5.5, every pair. @@ -2976,6 +3006,14 @@ package object query { private def conflict(a: Operand, b: Operand, path: Seq[String]): Conflict = if (a.branch <= b.branch) Conflict(a, b, path) else Conflict(b, a, path) + /** The type of one node at one position ([[typeNode]]): the operand that decides it, and + * whether DuckDB 1.5.5 gives the node NO type there -- at the position or at any field or + * element inside it -- which the rules answer as `VARCHAR`. + */ + private final case class NodeType(operand: Operand, untyped: Boolean) + + private def typed(operand: Operand): NodeType = NodeType(operand, untyped = false) + /** Nested operands -- lists, structs, points -- typed together, every operand of a node at * once, by the SAME rule as a column ([[typeNode]]): lists by their elements; structs and * points by their fields, merged as DuckDB 1.5.5 merges them (MEASURED): every field of every @@ -2985,16 +3023,23 @@ package object query { * beside a struct is refused, as a list's element or a field the rule refuses is -- an element * named by its lists. */ - private def nestedNode(operands: Seq[Operand], path: Seq[String]): Either[Conflict, Operand] = { + private def nestedNode( + operands: Seq[Operand], + path: Seq[String] + ): Either[Conflict, NodeType] = { val (lists, structs) = operands.partition(o => !structLike(o.folded)) if (lists.nonEmpty && structs.nonEmpty) Left(conflict(lists.head, structs.head, path)) else if (structs.isEmpty) { val elements = lists.map(o => o.at(o.valueType.element.getOrElse(SetOperationType(SQLTypes.Any)))) typeNode(elements, path) match { - case Right(e) => + case Right(NodeType(e, untyped)) => Right( - lists.head.at(SetOperationType(SQLTypes.Array(e.folded), element = Some(e.valueType))) + NodeType( + lists.head + .at(SetOperationType(SQLTypes.Array(e.folded), element = Some(e.valueType))), + untyped + ) ) case Left(Conflict(a, b, at)) if at == path => // a list's elements refused: the lists are named, with their types @@ -3002,25 +3047,31 @@ package object query { Left(conflict(list(a), list(b), path)) case Left(refused) => Left(refused) } - } else if (structs.forall(_.folded == SQLTypes.GeoPoint)) Right(structs.head) + } else if (structs.forall(_.folded == SQLTypes.GeoPoint)) Right(typed(structs.head)) // a STRUCT whose fields are not known has nothing to merge: the position's fields are not // known either else if (structs.exists(_.valueType.fields.isEmpty)) - Right(structs.head.at(SetOperationType(SQLTypes.Struct))) + Right(typed(structs.head.at(SetOperationType(SQLTypes.Struct)))) else { val names = structs.flatMap(_.valueType.fields.map(_.name)).foldLeft(Vector.empty[String]) { (seen, name) => if (seen.exists(_.equalsIgnoreCase(name))) seen else seen :+ name } names - .foldLeft[Either[Conflict, Vector[StructField]]](Right(Vector.empty)) { - case (Right(merged), name) => + .foldLeft[Either[Conflict, (Vector[StructField], Boolean)]]( + Right((Vector.empty, false)) + ) { + case (Right((merged, untyped)), name) => val having = structs.flatMap { o => o.valueType.fields.find(_.name.equalsIgnoreCase(name)).map(f => o.at(f.valueType)) } - typeNode(having, path :+ name).map(t => merged :+ StructField(name, t.valueType)) + typeNode(having, path :+ name).map(t => + (merged :+ StructField(name, t.operand.valueType), untyped || t.untyped) + ) case (refused, _) => refused } - .map(fields => structs.head.at(SetOperationType(SQLTypes.Struct, fields))) + .map { case (fields, untyped) => + NodeType(structs.head.at(SetOperationType(SQLTypes.Struct, fields)), untyped) + } } } @@ -3049,15 +3100,19 @@ package object query { * `BOOLEAN`, a `TIME` beside a date or a number -- every value then spelled as text. * * The type does not depend on the operands' order. A node whose type is not known (`ANY`) - * refuses nothing; a `NULL` literal takes the others' type. + * refuses nothing; a `NULL` literal takes the others' type. The answer says, beside the type, + * whether DuckDB gives the node none there, at the position or at any field or element inside + * ([[NodeType.untyped]]): every `VARCHAR` of the last rule where DuckDB gives none -- also a + * text beside a `VARBINARY` and another type, which DuckDB makes a `BLOB` it then cannot cast + * the other type to. */ - private def typeNode(operands: Seq[Operand], path: Seq[String]): Either[Conflict, Operand] = + private def typeNode(operands: Seq[Operand], path: Seq[String]): Either[Conflict, NodeType] = operands.find(_.folded == SQLTypes.Any) match { - case Some(unknown) => Right(unknown) + case Some(unknown) => Right(typed(unknown)) case None => val valued = operands.filterNot(_.folded == SQLTypes.Null) valued.headOption match { - case None => Right(operands.head) + case None => Right(typed(operands.head)) case Some(first) => valued.find(o => nested(o.folded)) match { case Some(list) => @@ -3073,8 +3128,16 @@ package object query { // beside any other mix is VARCHAR val others = valued.filterNot(_.folded == SQLTypes.Varchar) if (others.nonEmpty && others.forall(_.folded == SQLTypes.VarBinary)) - Right(others.head) - else Right(text) + Right(typed(others.head)) + else + // a BLOB beside another type besides the text: DuckDB gives the node no + // type (MEASURED), VARCHAR untyped + Right( + NodeType( + text, + untyped = others.exists(_.folded == SQLTypes.VarBinary) + ) + ) case None => val binary = valued.exists(_.folded == SQLTypes.VarBinary) ( @@ -3094,7 +3157,12 @@ package object query { ) ) commonType(valued) else None - Right(joined.getOrElse(first.at(SetOperationType(SQLTypes.Varchar)))) + // DuckDB gives the node no type: VARCHAR, the node UNTYPED + Right( + joined.fold( + NodeType(first.at(SetOperationType(SQLTypes.Varchar)), untyped = true) + )(typed) + ) } } } @@ -3156,7 +3224,8 @@ package object query { */ private sealed trait TypingNode private final case class TypingLeaf(branch: Int) extends TypingNode - private final case class TypingRun(children: Seq[TypingNode]) extends TypingNode + private final case class TypingRun(operator: SetOperator, children: Seq[TypingNode]) + extends TypingNode private def typingTree(node: SetOpNode): TypingNode = node match { @@ -3169,25 +3238,42 @@ package object query { flattened(l) ++ flattened(r) case other => Seq(typingTree(other)) } - TypingRun(flattened(left) ++ flattened(right)) + TypingRun(operator, flattened(left) ++ flattened(right)) } - /** The operand a node of the tree folds to at one column -- `None` when no branch under it - * declares one (`SELECT *`) -- or the two operands refused under it. + /** A node of the operator tree typed at one column: the operand it folds to -- `None` when no + * branch under it declares one (`SELECT *`) -- whether DuckDB gives it no type there + * ([[NodeType.untyped]]), and, for a run, each child typed the same way, in text order. + */ + private final case class TypedNode( + result: Option[Operand], + untyped: Boolean, + children: Seq[TypedNode] + ) + + /** A node of the tree typed at one column ([[TypedNode]]), or the two operands refused under + * it. */ private def typeOf( node: TypingNode, operand: Int => Option[Operand] - ): Either[Conflict, Option[Operand]] = + ): Either[Conflict, TypedNode] = node match { - case TypingLeaf(branch) => Right(operand(branch)) - case TypingRun(children) => + case TypingLeaf(branch) => Right(TypedNode(operand(branch), untyped = false, Nil)) + case TypingRun(_, children) => children - .foldLeft[Either[Conflict, Vector[Operand]]](Right(Vector.empty)) { - case (Right(typed), child) => typeOf(child, operand).map(typed ++ _) + .foldLeft[Either[Conflict, Vector[TypedNode]]](Right(Vector.empty)) { + case (Right(typed), child) => typeOf(child, operand).map(typed :+ _) case (conflict, _) => conflict } - .flatMap(typed => if (typed.isEmpty) Right(None) else typeNode(typed, Nil).map(Some(_))) + .flatMap { typedChildren => + val operands = typedChildren.flatMap(_.result) + if (operands.isEmpty) Right(TypedNode(None, untyped = false, typedChildren)) + else + typeNode(operands, Nil).map(t => + TypedNode(Some(t.operand), t.untyped, typedChildren) + ) + } } /** The type of each column of a set operation whose branches are joined by `operators`, or the @@ -3205,32 +3291,50 @@ package object query { private[elastic] def columnTypes( requests: Seq[SingleSearch], operators: Seq[SetOperator] - ): Either[String, Seq[SetOperationColumn]] = { - val joins = - if (operators.isEmpty) Seq.fill(math.max(requests.size - 1, 0))(UNION) else operators + ): Either[String, Seq[SetOperationColumn]] = + typedColumns(requests, operators).map(_.map(_._1)) + + /** The operators between the branches: `Nil` is `UNION ALL` everywhere. */ + private def joinsOf( + requests: Seq[SingleSearch], + operators: Seq[SetOperator] + ): Seq[SetOperator] = + if (operators.isEmpty) Seq.fill(math.max(requests.size - 1, 0))(UNION) else operators + + private def operatorCount(requests: Seq[SingleSearch], joins: Seq[SetOperator]): String = + s"A set operation over ${requests.size} branches needs ${requests.size - 1} operators, " + + s"got ${joins.size}" + + /** Each column of a set operation ([[SetOperationColumn]]) with the operator tree typed at it + * ([[TypedNode]]), or the column refused and why -- what [[columnTypes]] and [[columnTree]] + * both read, so the two cannot type a statement differently. + */ + private def typedColumns( + requests: Seq[SingleSearch], + operators: Seq[SetOperator] + ): Either[String, Seq[(SetOperationColumn, TypedNode)]] = { + val joins = joinsOf(requests, operators) val known = requests.map(declared).zipWithIndex.collect { case (Some(f), i) => (i, f) } if (known.isEmpty) Right(Nil) - else if (joins.size != requests.size - 1) - Left( - s"A set operation over ${requests.size} branches needs ${requests.size - 1} operators, " + - s"got ${joins.size}" - ) + else if (joins.size != requests.size - 1) Left(operatorCount(requests, joins)) else { val tree = typingTree(precedenceTree(requests.indices.toList, joins.toList)) val byBranch = known.toMap val width = known.map(_._2.size).min - (0 until width).foldLeft[Either[String, Vector[SetOperationColumn]]](Right(Vector.empty)) { + (0 until width).foldLeft[Either[String, Vector[(SetOperationColumn, TypedNode)]]]( + Right(Vector.empty) + ) { case (Right(columns), pos) => val leaves = requests.indices.map(branch => byBranch.get(branch).map(f => operandOf(branch, f(pos))) ) typeOf(tree, leaves) match { - case Right(result) => + case Right(typed) => Right( - columns :+ SetOperationColumn( - result.fold(SetOperationType(SQLTypes.Any))(r => resultOf(r.valueType)), + columns :+ (SetOperationColumn( + typed.result.fold(SetOperationType(SQLTypes.Any))(r => resultOf(r.valueType)), leaves.map(_.map(_.valueType)) - ) + ) -> typed) ) case Left(refused) => Left(refusal(pos, refused)) } @@ -3239,6 +3343,70 @@ package object query { } } + /** The NODES of a set operation whose branches are joined by `operators`, as DuckDB 1.5.5 types + * them -- the tree [[columnTypes]] types the statement over, its root the statement's -- or, + * `Left`, the refusal, the message of [[branchTypes]] word for word: `Left` exactly where + * [[branchTypes]] refuses. + * + * THE CONTRACT read by the relational engine (arrow, `app.softnetwork.elastic.arrow`), which + * types each node of the statement it runs and must give it the SAME type as core + * ([[SetOperationNode]]): the same inputs as [[branchTypes]] -- the branches in text order, + * each with its schema attached, and `MultiSearch.resolvedOperators` (`Nil` is `UNION ALL` + * everywhere); a run of one union kind is one node, every other operator a node over its two + * sides, `INTERSECT [ALL]` binding first. Each node gives each column the type the rules give + * it ([[typeNode]]), and says where DuckDB gives it none ([[SetOperationNode.untyped]]). The + * root's [[SetOperationNode.columns]] are the results of [[columnTypes]]. A set operation has + * two branches at least. + * + * Where no branch declares a column (every one a `SELECT *`), there is nothing to type, and + * [[branchTypes]] checks nothing, not even the number of operators: the nodes have no column, + * and an operator list of another length -- which `MultiSearch.validate()` refuses -- is read + * as `UNION ALL` past its end, an operator past the last branch ignored. + */ + private[elastic] def columnTree( + requests: Seq[SingleSearch], + operators: Seq[SetOperator] + ): Either[String, SetOperationNode] = { + val tooFew = s"A set operation needs two branches at least, got ${requests.size}" + if (requests.size < 2) Left(tooFew) + else + // `typedColumns` refuses what `branchTypes` refuses, the operator count included where a + // branch declares a column; elsewhere the tree has no column to type + typedColumns(requests, operators).flatMap { columns => + val joins = + joinsOf(requests, operators).padTo(requests.size - 1, UNION).take(requests.size - 1) + treeOperand( + typingTree(precedenceTree(requests.indices.toList, joins.toList)), + columns.map(_._2) + ) match { + case Right(root) => Right(root) + case Left(_) => Left(tooFew) + } + } + } + + /** A node of the tree as an operand of its parent: a branch's index, or the run as a + * [[SetOperationNode]], from the run typed at each column. + */ + private def treeOperand( + node: TypingNode, + typed: Seq[TypedNode] + ): Either[Int, SetOperationNode] = + node match { + case TypingLeaf(branch) => Left(branch) + case TypingRun(operator, children) => + Right( + SetOperationNode( + operator, + children.zipWithIndex.map { case (child, i) => + treeOperand(child, typed.map(_.children(i))) + }, + typed.map(_.result.fold(SetOperationType(SQLTypes.Any))(r => resultOf(r.valueType))), + typed.map(_.untyped) + ) + ) + } + /** Whether the branches of a set operation, joined by `operators`, may run, column by column, * once each branch's schema is attached -- `Right(())` -- or the refusal, by name. * diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/query/SetOperationColumnTreeSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/query/SetOperationColumnTreeSpec.scala new file mode 100644 index 00000000..b5af3203 --- /dev/null +++ b/sql/src/test/scala/app/softnetwork/elastic/sql/query/SetOperationColumnTreeSpec.scala @@ -0,0 +1,290 @@ +/* + * Copyright 2025 SOFTNETWORK + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package app.softnetwork.elastic.sql.query + +import app.softnetwork.elastic.sql.operator.{ + EXCEPT, + INTERSECT, + INTERSECT_ALL, + SetOperator, + UNION, + UNION_DISTINCT +} +import app.softnetwork.elastic.sql.parser.Parser +import app.softnetwork.elastic.sql.query.MultiSearch.{ + SetOperationNode, + SetOperationType, + StructField +} +import app.softnetwork.elastic.sql.schema.{Column, Table} +import app.softnetwork.elastic.sql.`type`.{SQLType, SQLTypes} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +/** A set operation's NODES, as DuckDB 1.5.5 types them ([[MultiSearch.columnTree]]): each node's + * operator and operands, the type it gives each column, and where DuckDB gives it NO type -- the + * nodes the relational engine must cast itself. Every expectation below is DuckDB 1.5.5's, RUN + * over `n = 5`, `x = 5.0`, `s = 'a'`. + */ +class SetOperationColumnTreeSpec extends AnyFlatSpec with Matchers { + + private val schema = Table( + "t", + columns = List( + Column("n", SQLTypes.Int), + Column("bg", SQLTypes.BigInt), + Column("x", SQLTypes.Double), + Column("s", SQLTypes.Keyword), + Column("bin", SQLTypes.VarBinary), + Column("d", SQLTypes.Date), + Column("ts", SQLTypes.Timestamp), + Column("t1", SQLTypes.Time), + Column("t2", SQLTypes.Time), + Column("oi", SQLTypes.Struct, multiFields = List(Column("a", SQLTypes.Int))), + Column( + "ot", + SQLTypes.Struct, + multiFields = List(Column("a", SQLTypes.Time), Column("c", SQLTypes.Keyword)) + ), + Column("ol", SQLTypes.Struct, multiFields = List(Column("a", SQLTypes.BigInt))), + Column("si", SQLTypes.Struct, multiFields = List(inner("p", SQLTypes.Int))), + Column("st", SQLTypes.Struct, multiFields = List(inner("p", SQLTypes.Time))), + Column("sl", SQLTypes.Struct, multiFields = List(inner("p", SQLTypes.BigInt))), + Column("ai", SQLTypes.Array(SQLTypes.Int)), + Column("atm", SQLTypes.Array(SQLTypes.Time)), + Column("geo", SQLTypes.GeoPoint) + ) + ) + + /** A field `name` that is an object with one field `c` of type `c`. */ + private def inner(name: String, c: SQLType): Column = + Column(name, SQLTypes.Struct, multiFields = List(Column("c", c))) + + private def resolved(sql: String): (Seq[SingleSearch], Seq[SetOperator]) = + Parser(sql) match { + case Right(m: MultiSearch) => (m.requests.map(_.update(Some(schema))), m.resolvedOperators) + case other => fail(s"[$sql] expected a set operation, got $other") + } + + private def tree(sql: String): SetOperationNode = { + val (requests, operators) = resolved(sql) + MultiSearch.columnTree(requests, operators) match { + case Right(root) => root + case Left(refusal) => fail(s"[$sql] refused: $refusal") + } + } + + private def select(item: String): String = s"SELECT $item AS k FROM t" + + private def scalar(t: SQLType): SetOperationType = SetOperationType(t) + + private def node( + operator: SetOperator, + operands: Seq[Either[Int, SetOperationNode]], + column: SetOperationType, + untyped: Boolean + ): SetOperationNode = SetOperationNode(operator, operands, Seq(column), Seq(untyped)) + + "A set operation's tree" should "type each node as DuckDB does: the contract's six cases" in { + // (n INTERSECT x) UNION s: INTERSECT is DOUBLE, then UNION VARCHAR -- typed: DuckDB answers + // {'5.0', 'a'}; casting n and x to VARCHAR first would answer {'a'} + tree(s"${select("n")} INTERSECT ${select("x")} UNION ${select("s")}") shouldBe node( + UNION_DISTINCT, + Seq( + Right(node(INTERSECT, Seq(Left(0), Left(1)), scalar(SQLTypes.Double), untyped = false)), + Left(2) + ), + scalar(SQLTypes.Varchar), + untyped = false + ) + // n UNION x UNION ALL s: a run of UNION, then UNION ALL over it -- DuckDB answers + // {'5.0', 'a'}; casting first would answer {'5', '5.0', 'a'} + tree(s"${select("n")} UNION ${select("x")} UNION ALL ${select("s")}") shouldBe node( + UNION, + Seq( + Right( + node(UNION_DISTINCT, Seq(Left(0), Left(1)), scalar(SQLTypes.Double), untyped = false) + ), + Left(2) + ), + scalar(SQLTypes.Varchar), + untyped = false + ) + // bin UNION d UNION n: one run DuckDB gives no type ("Unimplemented type for cast (DATE -> + // BLOB)"): VARCHAR, untyped + tree(s"${select("bin")} UNION ${select("d")} UNION ${select("n")}") shouldBe node( + UNION_DISTINCT, + Seq(Left(0), Left(1), Left(2)), + scalar(SQLTypes.Varchar), + untyped = true + ) + // CAST(ts AS TIME) UNION d: a TIME beside a DATE, untyped + tree(s"${select("CAST(ts AS TIME)")} UNION ${select("d")}") shouldBe node( + UNION_DISTINCT, + Seq(Left(0), Left(1)), + scalar(SQLTypes.Varchar), + untyped = true + ) + // (t1 INTERSECT t2) UNION ALL bin: INTERSECT typed TIME, then UNION ALL untyped + tree(s"${select("t1")} INTERSECT ${select("t2")} UNION ALL ${select("bin")}") shouldBe node( + UNION, + Seq( + Right(node(INTERSECT, Seq(Left(0), Left(1)), scalar(SQLTypes.Time), untyped = false)), + Left(2) + ), + scalar(SQLTypes.Varchar), + untyped = true + ) + // two structs whose field `a` is an INT and a TIME: `a` is VARCHAR, the column untyped, the + // struct merged in DuckDB's order + tree(s"${select("oi")} UNION ${select("ot")}") shouldBe node( + UNION_DISTINCT, + Seq(Left(0), Left(1)), + SetOperationType( + SQLTypes.Struct, + Seq(StructField("a", SQLTypes.Varchar), StructField("c", SQLTypes.Varchar)) + ), + untyped = true + ) + } + + it should "say where DuckDB gives no type at any depth: a field inside a field, an element" in { + // p.c an INT beside a TIME: untyped at the column + tree(s"${select("si")} UNION ALL ${select("st")}") shouldBe node( + UNION, + Seq(Left(0), Left(1)), + SetOperationType( + SQLTypes.Struct, + Seq( + StructField( + "p", + SetOperationType(SQLTypes.Struct, Seq(StructField("c", SQLTypes.Varchar))) + ) + ) + ), + untyped = true + ) + // p.c an INT beside a BIGINT: typed + tree(s"${select("si")} UNION ALL ${select("sl")}") shouldBe node( + UNION, + Seq(Left(0), Left(1)), + SetOperationType( + SQLTypes.Struct, + Seq( + StructField( + "p", + SetOperationType(SQLTypes.Struct, Seq(StructField("c", SQLTypes.BigInt))) + ) + ) + ), + untyped = false + ) + tree(s"${select("oi")} INTERSECT ALL ${select("ol")}").untyped shouldBe Seq(false) + // lists: their elements + val lists = tree(s"${select("ai")} EXCEPT ${select("atm")}") + lists.operator shouldBe EXCEPT + lists.columns shouldBe Seq( + SetOperationType(SQLTypes.Array(SQLTypes.Varchar), element = Some(scalar(SQLTypes.Varchar))) + ) + lists.untyped shouldBe Seq(true) + // a text beside a VARBINARY and a number: DuckDB makes a BLOB it cannot cast the number to + tree(s"${select("s")} UNION ${select("bin")} UNION ${select("n")}").untyped shouldBe Seq(true) + // a text beside a VARBINARY only: a BLOB, typed + val blob = tree(s"${select("s")} UNION ${select("bin")}") + blob.columns shouldBe Seq(scalar(SQLTypes.VarBinary)) + blob.untyped shouldBe Seq(false) + // points alone stay a point + tree(s"${select("geo")} UNION ${select("geo")}").columns shouldBe Seq(SetOperationType.GeoPoint) + } + + it should "type each column of a node, the root's being columnTypes' results" in { + val sql = + "SELECT n, bin, d AS e FROM t UNION SELECT x, d, ts FROM t INTERSECT SELECT bg, d, d FROM t" + val (requests, operators) = resolved(sql) + val root = tree(sql) + root.operator shouldBe UNION_DISTINCT + root.columns shouldBe Seq( + scalar(SQLTypes.Double), + scalar(SQLTypes.Varchar), + scalar(SQLTypes.Timestamp) + ) + root.untyped shouldBe Seq(false, true, false) + MultiSearch.columnTypes(requests, operators).map(_.map(_.result)) shouldBe Right(root.columns) + // the INTERSECT binds first: its own types + root.operands.head shouldBe Left(0) + root.operands(1) match { + case Right(intersect) => + intersect.operator shouldBe INTERSECT + intersect.operands shouldBe Seq(Left(1), Left(2)) + intersect.columns shouldBe Seq( + scalar(SQLTypes.Double), + scalar(SQLTypes.Date), + scalar(SQLTypes.Timestamp) + ) + intersect.untyped shouldBe Seq(false, false, false) + case other => fail(s"not a node: $other") + } + // a date field's mapping format is the branch's, never a node's + def formats(t: SetOperationType): Seq[String] = + t.format.toSeq ++ t.fields.flatMap(f => formats(f.valueType)) ++ + t.element.toSeq.flatMap(formats) + root.columns.flatMap(formats) shouldBe empty + } + + it should "keep a SELECT * branch as an operand, and type the others" in { + val root = tree(s"${select("n")} UNION ALL SELECT * FROM t UNION ALL ${select("x")}") + root.operands shouldBe Seq(Left(0), Left(1), Left(2)) + root.columns shouldBe Seq(scalar(SQLTypes.Double)) + // a node under which no branch declares a column: its type is not known + val star = tree(s"SELECT * FROM t INTERSECT SELECT * FROM t UNION ALL ${select("n")}") + star.operands.head match { + case Right(intersect) => intersect.columns shouldBe Seq(scalar(SQLTypes.Any)) + case other => fail(s"not a node: $other") + } + star.columns shouldBe Seq(scalar(SQLTypes.Int)) + } + + it should "refuse what branchTypes refuses, in its words" in { + Seq( + s"${select("d")} UNION ALL ${select("n")}", + s"${select("n")} EXCEPT ${select("d")} UNION ALL ${select("s")}", + s"${select("oi")} UNION ${select("n")}" + ).foreach { sql => + val (requests, operators) = resolved(sql) + val refusal = MultiSearch.branchTypes(requests, operators) + refusal.isLeft shouldBe true + MultiSearch.columnTree(requests, operators) shouldBe refusal + } + val (requests, _) = resolved(s"${select("n")} UNION ALL ${select("x")}") + MultiSearch.columnTree(requests, Seq(UNION, UNION)) shouldBe + Left("A set operation over 2 branches needs 1 operators, got 2") + MultiSearch.columnTree(requests.take(1), Nil) shouldBe + Left("A set operation needs two branches at least, got 1") + // no branch declares a column: branchTypes checks nothing, and neither does the tree + val (stars, _) = resolved("SELECT * FROM t UNION ALL SELECT * FROM t UNION ALL SELECT * FROM t") + Seq(Nil, Seq(UNION, UNION), Seq(INTERSECT), Seq(UNION, UNION, UNION)).foreach { ops => + MultiSearch.branchTypes(stars, ops) shouldBe Right(()) + MultiSearch.columnTree(stars, ops).map(_.columns) shouldBe Right(Nil) + } + // an operator list too short is read as UNION ALL past its end: INTERSECT binds first + MultiSearch.columnTree(stars, Seq(INTERSECT)).map(_.operands) shouldBe + Right(Seq(Right(SetOperationNode(INTERSECT, Seq(Left(0), Left(1)), Nil, Nil)), Left(2))) + MultiSearch.columnTree(requests, Nil).map(_.operator) shouldBe Right(UNION) + MultiSearch.columnTree(requests, Seq(INTERSECT_ALL)).map(_.operator) shouldBe + Right(INTERSECT_ALL) + } +} diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/query/SetOperationTypeRuleSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/query/SetOperationTypeRuleSpec.scala index b653391a..b0cbc852 100644 --- a/sql/src/test/scala/app/softnetwork/elastic/sql/query/SetOperationTypeRuleSpec.scala +++ b/sql/src/test/scala/app/softnetwork/elastic/sql/query/SetOperationTypeRuleSpec.scala @@ -371,4 +371,84 @@ class SetOperationTypeRuleSpec extends AnyFlatSpec with Matchers { } wrong shouldBe empty } + + /** The table's rule 2: where DuckDB 1.5.5 gives a node NO type and the rule answers `VARCHAR`. */ + private def untyped(set: Set[String]): Boolean = table.get(set).contains("VARCHAR" -> "FAILS") + + /** What differs between a node of [[MultiSearch.columnTree]] and the table, for a node joining + * `items` by `operator` -- with the type the rule gives it. A run of one union kind is ONE node + * over every item; `INTERSECT` and `EXCEPT` nodes are binary, left to right. + */ + private def nodeErrors( + node: MultiSearch.SetOperationNode, + operator: String, + items: Seq[String] + ): (Seq[String], String) = { + val (operands, set, subtree) = + if (operator.startsWith("UNION") || items.size == 2) + (items.indices.map(Left(_)), items.toSet, Nil) + else + node.operands.headOption match { + case Some(Right(sub)) => + val (errors, before) = nodeErrors(sub, operator, items.init) + (Seq(Right(sub), Left(items.size - 1)), Set(before, items.last), errors) + case other => (Nil, items.toSet, Seq(s"a binary node over ${items.size} items is $other")) + } + val errors = subtree ++ + (if (node.operator.sql == operator) Nil else Seq(s"operator ${node.operator.sql}")) ++ + (if (node.operands == operands) Nil else Seq(s"operands ${node.operands}")) ++ + (if (node.untyped == Seq(untyped(set))) Nil + else Seq(s"${set.mkString("+")} untyped ${node.untyped}")) + (errors, rule(set)) + } + + "Every node of a set operation" should "be typed as columnTypes types it, untyped where DuckDB gives none" in { + val sets = (types :+ "NULL").map(t => Seq(t, t)) ++ + (types :+ "NULL").combinations(2).flatMap(_.permutations) ++ + (types :+ "NULL").combinations(3).flatMap(_.permutations) + val statements = for { + (depth, prefix) <- depths + items <- sets + operator <- Seq("UNION ALL", "UNION", "INTERSECT", "EXCEPT") + } yield ( + depth, + operator, + items, + items + .map(i => s"SELECT ${if (i == "NULL") "NULL" else s"${prefix}_$i"} AS k FROM t") + .mkString(s" $operator ") + ) + statements.size shouldBe 13120 + var nodes = 0 + var untypedNodes = 0 + val wrong = statements.flatMap { case (depth, operator, items, sql) => + Parser(sql) match { + case Right(m: MultiSearch) => + val requests = m.requests.map(_.update(Some(schema))) + ( + MultiSearch.columnTree(requests, m.resolvedOperators), + MultiSearch.columnTypes(requests, m.resolvedOperators) + ) match { + case (Left(a), Left(b)) if a == b => Nil + case (Right(root), Right(columns)) => + def count(n: MultiSearch.SetOperationNode): Unit = { + nodes += 1 + if (n.untyped.exists(identity)) untypedNodes += 1 + n.operands.foreach(_.foreach(count)) + } + count(root) + (if (root.columns == columns.map(_.result)) Nil + else Seq(s"root ${root.columns} instead of ${columns.map(_.result)}")) ++ + nodeErrors(root, operator, items)._1 + .map(e => s"[$depth] [$sql] $e") + case (tree, types) => Seq(s"[$depth] [$sql] tree $tree, types $types") + } + case other => fail(s"[$sql] expected a set operation, got $other") + } + } + wrong shouldBe empty + // every node of every statement the rule answers, and those DuckDB gives no type -- counted + // from the table alone + (nodes, untypedNodes) shouldBe ((14016, 5136)) + } } diff --git a/testkit/src/main/scala/app/softnetwork/elastic/client/SetOperationTypeSpec.scala b/testkit/src/main/scala/app/softnetwork/elastic/client/SetOperationTypeSpec.scala index 2e476721..bf6589c8 100644 --- a/testkit/src/main/scala/app/softnetwork/elastic/client/SetOperationTypeSpec.scala +++ b/testkit/src/main/scala/app/softnetwork/elastic/client/SetOperationTypeSpec.scala @@ -608,6 +608,18 @@ trait SetOperationTypeSpec extends AnyFlatSpecLike with ElasticDockerTestKit wit */ private val formatDocs: Seq[String] = Seq("f1", "f2", "f3", "f4", "p1") + /** A date field core cannot read without guessing -- a region id, whose offset comes from zone + * data -- as a column `z` and an object's field `fz.a`, beside default-format partners: the + * pattern each major reads it with (MEASURED, core's date-format corpora: Joda's `ZZZ` on + * Elasticsearch 6, `VV` on 7 and 8). And an `epoch_second` field holding an array of arrays, + * which Elasticsearch accepts and flattens, as a column `v` and an object's field `fv.a`. + */ + private val zoneTable = "set_operation_zones" + + private val zoned = "2024-01-31 Europe/Paris" + + private def zoneFormat: String = if (esMajor == 6) "yyyy-MM-dd ZZZ" else "yyyy-MM-dd VV" + private def formatJson(id: String): String = (Seq("id" -> ("\"" + id + "\"")) ++ formatted.flatMap(f => @@ -673,6 +685,22 @@ trait SetOperationTypeSpec extends AnyFlatSpecLike with ElasticDockerTestKit wit ", PRIMARY KEY (id))" ) load(formatTable, formatDocs.map(formatJson).toList) + Try(Await.result(client.run(s"DROP TABLE IF EXISTS $zoneTable"), 60.seconds)) + run( + s"CREATE TABLE $zoneTable (id KEYWORD, z TIMESTAMP OPTIONS (format = '$zoneFormat'), " + + s"fz STRUCT FIELDS(a TIMESTAMP OPTIONS (format = '$zoneFormat')), ts TIMESTAMP, " + + "fts STRUCT FIELDS(a TIMESTAMP), d DATE, v TIMESTAMP OPTIONS (format = 'epoch_second'), " + + "fv STRUCT FIELDS(a TIMESTAMP OPTIONS (format = 'epoch_second')), PRIMARY KEY (id))" + ) + load( + zoneTable, + List( + s"""{"id":"z1","z":"$zoned","fz":{"a":"$zoned"},""" + + """"v":[[1706697000,1706697001]],"fv":{"a":[[1706697000,1706697001]]}}""", + """{"id":"p1","ts":"2024-02-01T00:00:00Z","fts":{"a":"2024-02-01T00:00:00Z"},""" + + """"d":"2024-02-01"}""" + ) + ) () } @@ -680,6 +708,7 @@ trait SetOperationTypeSpec extends AnyFlatSpecLike with ElasticDockerTestKit wit Try(Await.result(client.run(s"DROP TABLE IF EXISTS $table"), 60.seconds)) Try(Await.result(client.run(s"DROP TABLE IF EXISTS $textTable"), 60.seconds)) Try(Await.result(client.run(s"DROP TABLE IF EXISTS $formatTable"), 60.seconds)) + Try(Await.result(client.run(s"DROP TABLE IF EXISTS $zoneTable"), 60.seconds)) super.afterAll() } @@ -1629,6 +1658,145 @@ trait SetOperationTypeSpec extends AnyFlatSpecLike with ElasticDockerTestKit wit wrong shouldBe empty } + /** A date whose branches AGREE is read as the date Elasticsearch indexed when one of its branches + * is a field with a mapping format of its own -- seconds, custom patterns, a DATE's day -- in + * every leg, the default-format one too, at the column and at an object's field (every other + * entry of the object as stored): every value a date of its class, on every route. A date of the + * default format keeps its value as read. + */ + it should "read an agreeing date of a mapping format of its own, at every depth, on every route" in { + // the date at `depth` of a row's value: the column's, or the object's field `a` + def at(depth: String, k: Any): Any = + (depth, k) match { + case (_, null) => null + case ("column", v) => v + case ("field", m: scala.collection.Map[_, _]) => + m.asInstanceOf[scala.collection.Map[String, Any]].getOrElse("a", null) + case (_, other) => ("unexpected", other) + } + val wrong = for { + f <- formatted.filter(_.format.isDefined) + (depth, item, partner) <- Seq( + ("column", s"c_${f.key}", if (f.declared == "DATE") "pd" else "pts"), + ("field", s"f_${f.key}", if (f.declared == "DATE") "fd" else "fts") + ) + (route, limited) <- routes + sql = Seq(item, partner) + .map(i => s"SELECT $i AS k FROM $formatTable" + (if (limited) " LIMIT 100" else "")) + .mkString(" UNION ALL ") + error <- rowsOf(sql, route) match { + case Left(e) => Some(s"[$route] [$sql] failed: $e") + case Right(rows) => + def read(instant: Instant): Any = + if (f.declared == "DATE") instant.atZone(ZoneOffset.UTC).toLocalDate + else instant.atZone(ZoneOffset.UTC) + val partnerValue: Any = + if (f.declared == "DATE") LocalDate.parse("2024-02-01") + else ZonedDateTime.parse("2024-02-01T00:00:00Z") + val want = formatDocs.map(id => + f.values.get(id).fold[Any](null) { case (_, instant, byMajor) => + read(byMajor.getOrElse(esMajor, instant)) + } + ) ++ formatDocs.map(id => if (id == "p1") partnerValue else null) + val got = rows.map(r => at(depth, r.getOrElse("k", null))) + var unmatched = got.map(shown) + val missing = want.map(shown).filterNot { w => + unmatched.indexOf(w) match { + case -1 => false + case i => unmatched = unmatched.patch(i, Nil, 1); true + } + } + if (missing.isEmpty && unmatched.isEmpty) None + else + Some( + s"[$route] [$sql] missing ${missing.mkString(", ")} / unexpected " + + unmatched.mkString(", ") + ) + } + } yield error + // the control: dates of the default format keep their values as read -- the year 2024 written + // as a text and as a number -- at the column and at an object's field + val control = for { + (depth, item, partner) <- Seq(("column", "c_def", "pts"), ("field", "f_def", "fts")) + (route, limited) <- routes + sql = Seq(item, partner) + .map(i => s"SELECT $i AS k FROM $formatTable" + (if (limited) " LIMIT 100" else "")) + .mkString(" UNION ALL ") + error <- rowsOf(sql, route) match { + case Left(e) => Some(s"[$route] [$sql] failed: $e") + case Right(rows) => + val got = rows.map(r => shown(at(depth, r.getOrElse("k", null)))) + if (got.contains(("2024", "String")) && got.contains((2024, "Integer"))) None + else Some(s"[$route] [$sql] read ${got.mkString(", ")}") + } + } yield error + (wrong ++ control) shouldBe empty + } + + /** A date core cannot read without guessing -- a region id, whose offset comes from zone data -- + * and a value of several values -- an array of arrays -- where its branches AGREE are answered + * as stored, at the column and at an object's field, on every route, as they always were: no + * statement core answered fails. Where its branches disagree the date converts, and fails the + * statement, naming the value, the field and the format. + */ + it should "keep an agreeing date it cannot read as stored, and fail it where it converts" in { + def sql(left: String, right: String, limited: Boolean): String = + Seq(left, right) + .map(i => s"SELECT $i AS k FROM $zoneTable" + (if (limited) " LIMIT 100" else "")) + .mkString(" UNION ALL ") + val kept = for { + (left, right, stored, partner) <- Seq( + ("z", "ts", zoned: Any, ZonedDateTime.parse("2024-02-01T00:00:00Z"): Any), + ( + "fz", + "fts", + ListMap("a" -> zoned): Any, + ListMap("a" -> ZonedDateTime.parse("2024-02-01T00:00:00Z")): Any + ), + // an array of arrays: several values, as stored + ( + "v", + "ts", + List(List(1706697000, 1706697001)): Any, + ZonedDateTime.parse("2024-02-01T00:00:00Z"): Any + ), + ( + "fv", + "fts", + ListMap("a" -> List(List(1706697000, 1706697001))): Any, + ListMap("a" -> ZonedDateTime.parse("2024-02-01T00:00:00Z")): Any + ) + ) + (route, limited) <- routes + statement = sql(left, right, limited) + error <- rowsOf(statement, route) match { + case Left(e) => Some(s"[$route] [$statement] failed: $e") + case Right(rows) => + def norm(v: Any): Any = v match { + case z: ZonedDateTime => z.toInstant + case m: scala.collection.Map[_, _] => + m.map { case (k, e) => String.valueOf(k) -> norm(e) }.toMap + case other => other + } + val got = rows.map(r => norm(r.getOrElse("k", null))).sortBy(String.valueOf) + val want = Seq(stored, null, null, partner).map(norm).sortBy(String.valueOf) + if (got == want) None else Some(s"[$route] [$statement] answered $got instead of $want") + } + } yield error + val failed = for { + (route, limited) <- routes + statement = sql("z", "d", limited) + error <- rowsOf(statement, route) match { + case Left(msg) + if msg.contains(s"Could not read '$zoned' as a TIMESTAMP") && + msg.contains(s"the date format '$zoneFormat' of field z") => + None + case other => Some(s"[$route] [$statement] answered $other instead of failing") + } + } yield error + (kept ++ failed) shouldBe empty + } + it should "refuse by name only a date beside a number, before anything runs" in { val wrong = refused.flatMap { case (sql, reason) => rowsOf(sql, "run") match {