diff --git a/core/src/main/scala/app/softnetwork/elastic/client/DateRoute.scala b/core/src/main/scala/app/softnetwork/elastic/client/DateRoute.scala new file mode 100644 index 00000000..1a401d33 --- /dev/null +++ b/core/src/main/scala/app/softnetwork/elastic/client/DateRoute.scala @@ -0,0 +1,72 @@ +/* + * 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.`type`.{SQLType, SQLTypes} + +/** How ONE date position of a set operation's leg is read, decided BEFORE any value: from the type + * its branch declares, its date field's mapping `format` and the Elasticsearch major the statement + * runs on -- never from a value, so the answer cannot depend on the order the rows come in. + * + * - [[DateRoute.Typed]]: the declared type, every row -- a date read by its format, as that + * major indexed it, or a computed value. A value core cannot read there fails the statement, + * at any row (`SetOperationValues.Unreadable.Fail`). + * - [[DateRoute.AsStored]]: `VARCHAR`, the value as stored, every row -- where core cannot read + * every value of the format for sure, which the format alone says + * ([[MappedDateFormat.unread]]), or where the major is not known. + */ +private[client] sealed trait DateRoute + +private[client] object DateRoute { + + /** Read by its format, or as a computed value: the declared type, every row. */ + case object Typed extends DateRoute + + /** `VARCHAR` as stored, every row, decided before any row; `reason` names the format and why. */ + final case class AsStored(reason: String) extends DateRoute + + /** `Typed` for a type that is not a date read by its format -- the positions + * `SetOperationValues.readAs` reads with one: a `DATE`, a `TIMESTAMP` or a `TIME` that has a + * format, Elasticsearch's default included -- else by the major `dates` names (its `dialect`, + * asked here, once per statement) and what the format reads ([[MappedDateFormat.unread]]): + * `AsStored` when that major is not known, or when core does not read every value of the format + * for sure. + */ + def of( + sqlType: SQLType, + format: Option[String], + dates: SetOperationValues.DateReading + ): DateRoute = + (sqlType, format) match { + case (SQLTypes.Date | SQLTypes.Timestamp | SQLTypes.Time, Some(spec)) => + dates.dialect match { + case None => + AsStored( + s"the date format '$spec' is read as the Elasticsearch version does, " + + dates.elasticVersion.fold("which is not known")(v => s"and core reads dates of no $v") + ) + case Some(dialect) => + MappedDateFormat(spec, dialect, dates.clock).unread.fold[DateRoute](Typed) { why => + AsStored( + s"core cannot read every value of the date format '$spec' as ${dialect.label} " + + s"does: $why" + ) + } + } + case _ => Typed + } +} diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala index 4d4cb118..d2c88565 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala @@ -91,6 +91,13 @@ trait ElasticClientDelegator extends ElasticClientApi with BulkTypes { override private[client] def executeVersion(): ElasticResult[String] = delegate.executeVersion() + override private[client] def buildElasticMajor: Option[Int] = + delegate.buildElasticMajor + + /** The delegate's: every wrapper of a client reads dates as ONE version, asked once. */ + override private[elastic] lazy val dateDialectVersion: Option[String] = + delegate.dateDialectVersion + /** Get Elasticsearch cluster name. * * @return diff --git a/core/src/main/scala/app/softnetwork/elastic/client/JavaDatePattern.scala b/core/src/main/scala/app/softnetwork/elastic/client/JavaDatePattern.scala index c20625f5..4afb044a 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/JavaDatePattern.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/JavaDatePattern.scala @@ -643,6 +643,25 @@ private[client] object JavaDatePattern { /** Why the pattern is not read, if it is not. */ def unread: Option[String] = compiled.left.toOption + /** The kind of its zone letter, when it reads zone data: `"id"` (`VV`) or `"name"` (`z`, `v`). + */ + override private[client] def zoneData: Option[String] = + compiled.toOption.map(_.zone).filter(kind => kind == "id" || kind == "name") + + /** On a major resolving SMARTLY (Elasticsearch 6.8's `8`-prefixed mode), why the pattern reads + * an AM/PM with no hour, decided from ITS letters (quoted text aside) before any value -- + * `None` where it does not. The rule is the one of [[shape]], applied to the two classes it + * reads there: the AM/PM and the hour, any hour letter (`h`, `K`, `k`, `H`) standing as `h` + * (the measured `ah`). [[shape]] applies it to the letters a text carried, which a pattern + * read in a lexical pass (weeks of a week-based year on 6.8: `YYYY-ww a`) meets only value by + * value. + */ + private[client] lazy val amPmWithoutHour: Option[String] = + tokens(pattern).toOption.flatMap { ts => + val classes = ts.collect { case Letters(c, n) => letterClass(c, n) }.toSet + shape(classes.collect { case 'a' => 'a'; case 'h' | 'K' | 'k' | 'H' => 'h' }, "") + } + def parse(text: String): Parse = compiled match { case Left(reason) => Unmeasured(s"'$pattern' is not a date format core reads: $reason") diff --git a/core/src/main/scala/app/softnetwork/elastic/client/JodaDatePattern.scala b/core/src/main/scala/app/softnetwork/elastic/client/JodaDatePattern.scala index 33da57f4..591fd1e0 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/JodaDatePattern.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/JodaDatePattern.scala @@ -544,6 +544,10 @@ private[client] object JodaDatePattern { /** Why the pattern is not read, if it is not. */ def unread: Option[String] = compiled.left.toOption + /** The kind of its zone letter, when it reads zone data: `"id"` (`ZZZ`) or `"name"` (`z`). */ + override private[client] def zoneData: Option[String] = + compiled.toOption.map(_._2).filter(kind => kind == "id" || kind == "name") + def parse(text: String): Parse = compiled match { case Left(reason) => Unmeasured(s"'$pattern' is not a date format core reads: $reason") diff --git a/core/src/main/scala/app/softnetwork/elastic/client/MappedDateFormat.scala b/core/src/main/scala/app/softnetwork/elastic/client/MappedDateFormat.scala index 8335d7fd..5d40f03f 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/MappedDateFormat.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/MappedDateFormat.scala @@ -232,6 +232,35 @@ private[client] final case class MappedDateFormat( private def parses(text: String): Boolean = alternatives.exists(_.parse(text) != NotParsed) + + /** Why core does not read EVERY value of the format for sure, decided from the format alone, + * before any value -- `None` when every value it reads, it reads for sure: an alternative core + * does not read on this major ([[Unknown]]: a letter, a width or a combination no measurement + * covers, a name of Elasticsearch 6.8's `8`-prefixed mode); a custom pattern whose zone letter + * reads zone names or region ids ([[Alternative.zoneData]]), whose offset comes from the zone + * data of the JVM reading it, at any row; a pattern of Elasticsearch 6.8's `8`-prefixed mode + * reading an AM/PM with no hour ([[JavaDatePattern.Pattern.amPmWithoutHour]]). The first such + * alternative names the reason. + */ + private[client] lazy val unread: Option[String] = + alternatives.iterator.map(unreadBy).collectFirst { case Some(reason) => reason } + + private def unreadBy(alternative: Alternative): Option[String] = + alternative match { + case unknown: Unknown => + Some(unknown.parse(spec) match { + case Unmeasured(reason) => reason + case _ => s"'${unknown.name}' is not a date format core reads" + }) + case pattern: JavaDatePattern.Pattern if pattern.amPmWithoutHour.isDefined => + pattern.amPmWithoutHour + case other => + other.zoneData.map { kind => + s"its zone letter reads ${if (kind == "id") "region ids" else "zone names"}, 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" + } + } } private[client] object MappedDateFormat { @@ -286,6 +315,12 @@ private[client] object MappedDateFormat { /** A name core reads, as opposed to a custom pattern or an unknown name. */ def isName: Boolean = true + + /** The zone data its zone letter reads, decided from the format alone: `"name"` (a zone name: + * `java.time`'s `z` and `v`, Joda-Time's `z` and `zzzz`) or `"id"` (a region id: `java.time`'s + * `VV`, Joda-Time's `ZZZ`) -- `None` for no zone, or an offset. + */ + private[client] def zoneData: Option[String] = None } /** A name or a pattern core does not read on this major: a value it decides is undecidable. */ diff --git a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala index dbff9581..624c0dc9 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala @@ -2458,14 +2458,16 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC } } - /** The Elasticsearch version the statement runs on (`VersionApi.version`, cached per client): a - * set operation reads a date field's values as THAT major reads the field's mapping format - * (`SetOperationValues.converters`, which asks for it at most once per statement, and only when - * a converting position reads a date field). `None` for a client that cannot tell it. + /** The Elasticsearch version the statement reads dates as (`VersionApi.dateDialectVersion`, asked + * once per client): a set operation reads a date field's values as THAT major reads the field's + * mapping format (`SetOperationValues.converters`, which asks for it at most once per statement, + * and only when a position reads a date field: a converting position, or one whose branches + * agree and hold a date field with a mapping format of its own). For a user denied `GET /`, the + * major the client module is built for; `None` for a client that can tell neither. */ private def setOperationVersion: Option[String] = this match { - case api: VersionApi => api.version.toOption + case api: VersionApi => api.dateDialectVersion case _ => None } 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 af8cfb77..fd997775 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/SetOperationValues.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/SetOperationValues.scala @@ -76,7 +76,9 @@ private[elastic] object SetOperationValues { * * `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. + * asked at most ONCE per statement, and only when a position reads a date field: a converting + * position, or one whose branches agree and hold a date field with a mapping format of its own + * ([[readsMappedDates]]). */ private[client] def converters( multiple: MultiSearch, @@ -160,6 +162,174 @@ private[elastic] object SetOperationValues { } } + /** ONE column of ONE leg of a set operation, typed BEFORE any value is read ([[legSchemas]]). + * + * @param sqlType + * the type the leg's values have there, at every depth: the branch's declared type, a date + * position whose format core cannot read for sure `VARCHAR` ([[DateRoute.AsStored]]) + * @param read + * the reader of one raw value there, answering `sqlType`'s class ([[legReaders]]' classes; + * `VARCHAR` a `String`, the text as stored) + * @param asStored + * the date positions answered as stored -- their path (`tsz`, `oz.z`) and why + */ + private[elastic] final case class LegSlot( + sqlType: SetOperationType, + read: Any => Any, + asStored: Seq[LegSlot.AsStored] + ) + + private[elastic] object LegSlot { + + /** A date position answered as stored: its `path` -- the field, `.` before a struct's field -- + * and the `reason`, naming its format. + */ + final case class AsStored(path: String, reason: String) + } + + /** A leg's slots, by position; `None` at a position whose values are typed as they come. */ + private[elastic] final case class LegSchema(slots: IndexedSeq[Option[LegSlot]]) + + /** THE SCHEMA of each leg of a set operation, decided BEFORE any value is read -- in leg order, + * `None` for a leg that declares no projection (`SELECT *`) -- or, `Left`, the statement's + * refusal (`MultiSearch.branchTypes`' message word for word). Same inputs as [[legReaders]]; the + * version is asked at most ONCE, and only when a position holds a date read by its format (the + * default one too). + * + * Each position of a leg is a [[LegSlot]] -- the leg's type there at every depth and the reader + * of its values -- or `None` where its values are typed as they come: + * + * - A date read by its mapping format is routed before any row ([[DateRoute.of]]): read as the + * branch declares it, as [[legReaders]] reads it with [[Unreadable.Fail]] -- a value core + * cannot read fails, at any row; or, where core cannot read every value of the format for + * sure (a zone name or a region id, a letter no measurement covers) or the major is not + * known, `VARCHAR`, the text as stored, every row -- at the column, in a struct's field and + * in a list's element, where the branches agree or not. The routed type is read by the SAME + * reader as every other position ([[legReader]]): a `VARCHAR` position is the text leg's. + * - `None` where [[legReader]] reads nothing (a column whose type is not known, `ANY` -- a + * `WITH` or a derived table aliasing the column --, a `NULL` literal); where the leg's OR + * the column's type holds, at any depth, a type not known (`ANY`, a `STRUCT` whose fields + * are not known) ([[unknownInside]]); and where the leg's type holds, at any depth, one read + * as it comes until the engine types it (`NUMERIC`, `TEMPORAL`, `VARBINARY`, `REAL`) + * ([[valueTyped]]). + * + * Built ONCE per statement; no decision is made per row. + */ + private[elastic] def legSchemas( + requests: Seq[SingleSearch], + operators: Seq[SetOperator], + version: => Option[String] + ): Either[String, Seq[Option[LegSchema]]] = + MultiSearch.columnTypes(requests, operators).map { columns => + val names = columnNames(requests, columns.size) + val dates = DateReading(version) + requests.indices.map { leg => + val search = requests(leg) + // an opaque `SELECT *` -- a projection of another width, or the branch + // `MultiSearch.columnTypes` types at no column -- has no position to type + if ( + search.select.fieldsWithComputedAliases.size != columns.size || + columns.forall(_.branches.lift(leg).flatten.isEmpty) + ) None + else + Some(LegSchema(columns.indices.map { position => + legSlot(columns(position), leg, names(position), fieldOf(search, position), dates) + })) + } + } + + /** Leg `index`'s slot at ONE column ([[legSchemas]]): its branch's type routed ([[routed]]), read + * by [[legReader]] over the column whose branch `index` is the routed type -- `None` where + * [[legReader]] reads nothing, where the branch's type is read as it comes ([[valueTyped]]), or + * where the column's type is not known at some depth ([[unknownInside]]). + */ + private[client] def legSlot( + column: SetOperationColumn, + index: Int, + name: String, + field: String, + dates: DateReading + ): Option[LegSlot] = + column.branches + .lift(index) + .flatten + .filterNot(branch => valueTyped(branch) || unknownInside(column.result)) + // whether the slot exists is [[legReader]]'s to say over the column AS DECLARED, before any + // date is routed (routing asks the version): routing makes a date `VARCHAR`, never `NULL`, + // and leaves the column's type alone, so [[legReader]] answers the same over the routed one + .filter(_ => Option(legReader(column, index, name, field, dates, Unreadable.Fail)).isDefined) + .flatMap { branch => + val (slotType, asStored) = routed(branch, if (field.isEmpty) name else field, dates) + Option( + legReader( + column.copy(branches = column.branches.updated(index, Some(slotType))), + index, + name, + field, + dates, + Unreadable.Fail + ) + ).map(LegSlot(slotType, _, asStored)) + } + + /** A branch's type with every date position [[DateRoute.of]] answers as stored made `VARCHAR`, at + * every depth -- a struct's field (`path.field`), a list's element (the list's path) -- and + * those positions. + */ + private def routed( + t: SetOperationType, + path: String, + dates: DateReading + ): (SetOperationType, Seq[LegSlot.AsStored]) = + t.sqlType match { + case SQLTypes.Struct => + val fields = t.fields.map { f => + val (valueType, asStored) = routed(f.valueType, s"$path.${f.name}", dates) + (f.copy(valueType = valueType), asStored) + } + (t.copy(fields = fields.map(_._1)), fields.flatMap(_._2)) + case SQLTypes.Array(_) => + t.element.fold((t, Seq.empty[LegSlot.AsStored])) { e => + val (element, asStored) = routed(e, path, dates) + (t.copy(sqlType = SQLTypes.Array(element.sqlType), element = Some(element)), asStored) + } + case _ => + DateRoute.of(t.sqlType, t.format, dates) match { + case DateRoute.Typed => (t, Nil) + case DateRoute.AsStored(reason) => + (SetOperationType(SQLTypes.Varchar), Seq(LegSlot.AsStored(path, reason))) + } + } + + /** Whether a leg's type holds, at any depth, a type whose values are typed as they come: one not + * known ([[unknownInside]]), or one the relational engine types from its values until it reads + * it (`NUMERIC`, `TEMPORAL`, `VARBINARY`, `REAL`). + */ + private def valueTyped(t: SetOperationType): Boolean = + holds(t) { + case SQLTypes.Any | SQLTypes.Numeric | SQLTypes.Temporal | SQLTypes.VarBinary | + SQLTypes.Real => + true + case _ => false + } + + /** Whether a type holds, at any depth, a type not known: `ANY`, or a `STRUCT` whose fields are + * not known. Where a COLUMN's type does, [[legReader]] reads every leg's value there as it + * comes, not as the leg's own type, which another leg left unknown there: no leg has a slot. + */ + private def unknownInside(t: SetOperationType): Boolean = + holds(t)(_ == SQLTypes.Any) + + /** Whether `t` holds, at the column or in a struct's field or a list's element at any depth, a + * scalar `leaf` answers for, or a `STRUCT` whose fields are not known. + */ + private def holds(t: SetOperationType)(leaf: SQLType => Boolean): Boolean = + t.sqlType match { + case SQLTypes.Struct => t.fields.isEmpty || t.fields.exists(f => holds(f.valueType)(leaf)) + case SQLTypes.Array(_) => t.element.forall(e => holds(e)(leaf)) + case scalar => leaf(scalar) + } + /** 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 @@ -213,8 +383,9 @@ private[elastic] object SetOperationValues { 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. + * mapping format ([[MappedDateFormat.Dialect]]). The version is asked at most ONCE per + * statement, and only when a position reads a date field -- which positions do, each caller + * says: [[converters]], [[legReaders]], [[legSchemas]]. */ private[client] final class DateReading( version: => Option[String], diff --git a/core/src/main/scala/app/softnetwork/elastic/client/VersionApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/VersionApi.scala index c1c9153e..53f5ac77 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/VersionApi.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/VersionApi.scala @@ -50,6 +50,20 @@ trait VersionApi extends ElasticClientHelpers { } } + /** The major of Elasticsearch the client module is built for (its build info's `elasticMajor`), + * by which dates are read when the cluster's version cannot be ([[dateDialectVersion]]); `None` + * in core, which builds for no major. + */ + private[client] def buildElasticMajor: Option[Int] = None + + /** The Elasticsearch version a statement reads dates as (`SetOperationValues`): the cluster's + * ([[version]]), else -- a user whose role is denied `GET /` -- the major the client module is + * built for ([[buildElasticMajor]]), else `None`. Asked ONCE per client, and kept: a privilege + * granted later is seen after the client restarts. + */ + private[elastic] lazy val dateDialectVersion: Option[String] = + version.toOption.orElse(buildElasticMajor.map(_.toString)) + // def checkVersion(check: Int => Boolean) // ======================================================================== diff --git a/core/src/test/scala/app/softnetwork/elastic/arrow/LegSchemasConsumerSmokeSpec.scala b/core/src/test/scala/app/softnetwork/elastic/arrow/LegSchemasConsumerSmokeSpec.scala new file mode 100644 index 00000000..df8176a5 --- /dev/null +++ b/core/src/test/scala/app/softnetwork/elastic/arrow/LegSchemasConsumerSmokeSpec.scala @@ -0,0 +1,84 @@ +/* + * 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.arrow + +import app.softnetwork.elastic.client.{GatewayApi, VersionApi, VersionDouble} +import app.softnetwork.elastic.client.SetOperationValues +import app.softnetwork.elastic.client.SetOperationValues.{LegSchema, LegSlot} +import app.softnetwork.elastic.client.metrics.{MetricsCollector, MetricsElasticClient} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +/** The leg schema API as the relational engine (`app.softnetwork.elastic.arrow`) reaches it: from a + * `GatewayApi`-typed client wrapped as its factory wraps it, the version read through + * `VersionApi.dateDialectVersion`, each leg's slots read through `LegSchema` and `LegSlot`. + */ +class LegSchemasConsumerSmokeSpec extends AnyFlatSpec with Matchers { + + /** A client denied `GET /`, built for Elasticsearch 8, behind the metrics wrapper. */ + private val gateway: GatewayApi = + new MetricsElasticClient(new VersionDouble(Some(8)), new MetricsCollector) + + private def legs(sql: String): Seq[Option[LegSchema]] = { + val (requests, operators) = VersionDouble.resolved(sql) + SetOperationValues.legSchemas( + requests, + operators, + gateway.asInstanceOf[VersionApi].dateDialectVersion + ) match { + case Right(schemas) => schemas + case Left(refusal) => fail(s"[$sql] refused: $refusal") + } + } + + /** Each leg's slot types, `None` where a position is typed as its values come. */ + private def slotTypes(sql: String): Seq[Seq[Option[String]]] = + legs(sql).map { leg => + leg.toSeq.flatMap(_.slots.map((slot: Option[LegSlot]) => slot.map(_.sqlType.sqlType.typeId))) + } + + "The leg schemas" should "be read by the relational engine on every form it runs" in { + // plain + slotTypes("SELECT c_es AS k, n FROM t UNION SELECT tsz AS k, x AS n FROM t") shouldBe + Seq(Seq(Some("TIMESTAMP"), Some("INT")), Seq(Some("VARCHAR"), Some("DOUBLE"))) + // multi-step + slotTypes( + "SELECT c_es AS k FROM t UNION SELECT ts AS k FROM t INTERSECT SELECT tsz AS k FROM t " + + "EXCEPT SELECT c_c1 AS k FROM t" + ) shouldBe Seq( + Seq(Some("TIMESTAMP")), + Seq(Some("TIMESTAMP")), + Seq(Some("VARCHAR")), + Seq(Some("TIMESTAMP")) + ) + // a WITH passing its column through: the plain column's slot + slotTypes( + "WITH a AS (SELECT tsz FROM t) SELECT tsz AS k FROM a UNION SELECT ts AS k FROM t" + ) shouldBe Seq(Seq(Some("VARCHAR")), Seq(Some("TIMESTAMP"))) + // a WITH aliasing in its CTE, a derived table: the column is ANY, no slot in any leg + Seq( + "WITH a AS (SELECT tsz AS k FROM t) SELECT k FROM a UNION SELECT ts AS k FROM t", + "SELECT k FROM (SELECT tsz AS k FROM t) dd UNION SELECT ts AS k FROM t" + ).foreach(sql => slotTypes(sql) shouldBe Seq(Seq(None), Seq(None))) + // the as-stored positions and the reader, as the engine reads them + val tsz: LegSlot = legs("SELECT tsz AS k FROM t UNION SELECT ts AS k FROM t").head + .flatMap(_.slots.head) + .getOrElse(fail("no slot")) + tsz.asStored.map((p: LegSlot.AsStored) => p.path) shouldBe Seq("tsz") + tsz.read("2024-01-31 10:30 CET") shouldBe "2024-01-31 10:30 CET" + } +} diff --git a/core/src/test/scala/app/softnetwork/elastic/client/DateDialectVersionSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/DateDialectVersionSpec.scala new file mode 100644 index 00000000..90f86294 --- /dev/null +++ b/core/src/test/scala/app/softnetwork/elastic/client/DateDialectVersionSpec.scala @@ -0,0 +1,284 @@ +/* + * 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.client.metrics.{MetricsCollector, MetricsElasticClient} +import app.softnetwork.elastic.client.result.{ + ElasticError, + ElasticFailure, + ElasticResult, + ElasticSuccess +} +import app.softnetwork.elastic.sql.{StringValue, Value} +import app.softnetwork.elastic.sql.parser.Parser +import app.softnetwork.elastic.sql.operator.SetOperator +import app.softnetwork.elastic.sql.query.{ + MultiSearch, + SQLAggregation, + SelectStatement, + SingleSearch +} +import app.softnetwork.elastic.sql.schema.{Column, Schema, Table} +import app.softnetwork.elastic.sql.serialization.JacksonConfig +import app.softnetwork.elastic.sql.`type`.SQLTypes +import ch.qos.logback.classic.{Level, Logger => LogbackLogger} +import ch.qos.logback.classic.spi.ILoggingEvent +import ch.qos.logback.core.read.ListAppender +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.slf4j.{Logger, LoggerFactory} + +import java.time.ZonedDateTime +import java.util.concurrent.atomic.AtomicInteger +import scala.collection.immutable.ListMap +import scala.jdk.CollectionConverters._ + +/** A client whose `GET /` answers `answer` -- a 403 by default, a role denied it -- built for the + * Elasticsearch `major` its module's build info says; it counts its `GET /` and keeps the leg + * projections a one-shot `_msearch` would carry, which it does not send. + */ +private[elastic] class VersionDouble(major: Option[Int], val log: Logger = VersionDouble.log) + extends NopeClientApi { + + /** What `GET /` answers: `None`, a 403. */ + @volatile var answer: Option[String] = None + + val calls = new AtomicInteger(0) + + @volatile var legProjections: Seq[LegProjection] = Nil + + override protected def logger: Logger = log + + override private[client] def buildElasticMajor: Option[Int] = major + + override private[client] def executeVersion(): ElasticResult[String] = { + calls.incrementAndGet() + answer.fold[ElasticResult[String]]( + ElasticFailure( + ElasticError("403 Forbidden: action [cluster:monitor/main]", statusCode = Some(403)) + ) + )(ElasticSuccess(_)) + } + + override def multiSearch( + elasticQueries: ElasticQueries, + fieldAliases: ListMap[String, String], + aggregations: ListMap[String, SQLAggregation], + fields: Seq[String] = Seq.empty, + nestedHits: Map[String, Seq[(String, String)]] = Map.empty, + rowInvariants: Seq[ListMap[String, Any]] = Seq.empty, + legProjections: Seq[LegProjection] = Seq.empty + )(implicit context: ConversionContext): ElasticResult[ElasticResponse] = { + this.legProjections = legProjections + ElasticFailure(ElasticError("not sent")) + } +} + +private[elastic] object VersionDouble { + val log: Logger = LoggerFactory.getLogger("app.softnetwork.elastic.client.VersionDouble") + + private def format(f: String): ListMap[String, Value[_]] = + ListMap[String, Value[_]]("format" -> StringValue(f)) + + /** The schema the doubles read set operations with. */ + val schema: 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("tsz", SQLTypes.Timestamp, options = format("yyyy-MM-dd HH:mm z")), + Column("ts", SQLTypes.Timestamp), + Column("d", SQLTypes.Date), + Column("n", SQLTypes.Int), + Column("x", SQLTypes.Double), + Column("s", SQLTypes.Keyword) + ) + ) + + /** A set operation's requests, with the schema attached, and its operators. */ + 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 => throw new IllegalArgumentException(s"[$sql] $other") + } +} + +/** The version a statement reads its dates as (`VersionApi.dateDialectVersion`): the cluster's, + * else -- a user whose role is denied `GET /` -- the major the client module is built for, asked + * ONCE per client, shared by every wrapper of it, read by the relational engine's leg schemas and + * by core's own `UNION ALL`. + */ +class DateDialectVersionSpec extends AnyFlatSpec with Matchers { + + implicit val context: ConversionContext = NativeContext + + private def received(json: String): Any = + ElasticConversion.jsonNodeToAny(JacksonConfig.objectMapper.readTree(json), ListMap.empty) + + 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" + + private def utc(s: String): ZonedDateTime = ZonedDateTime.parse(s) + + private def shown(v: Any): (Any, Class[_]) = + v match { + case null => (null, classOf[Null]) + case other => (other, other.getClass) + } + + private def forbidden(major: Option[Int] = Some(8)): VersionDouble = new VersionDouble(major) + + private def answering(version: String, major: Option[Int] = Some(8)): VersionDouble = { + val double = new VersionDouble(major) + double.answer = Some(version) + double + } + + /** Leg `leg`'s slot at the first column, as the relational engine reads it. */ + private def slot(sql: String, version: => Option[String], leg: Int = 0) = { + val (requests, operators) = VersionDouble.resolved(sql) + SetOperationValues.legSchemas(requests, operators, version) match { + case Right(legs) => legs(leg).flatMap(_.slots.head).getOrElse(fail(s"[$sql] no slot")) + case Left(refusal) => fail(s"[$sql] refused: $refusal") + } + } + + "A client denied GET /" should "read dates as the major its module is built for" in { + Seq(6, 7, 8).foreach { major => + val client = forbidden(Some(major)) + client.dateDialectVersion shouldBe Some(major.toString) + val typed = slot(union("c_es", "ts", "UNION"), client.dateDialectVersion) + typed.sqlType.sqlType shouldBe SQLTypes.Timestamp + typed.asStored shouldBe empty + shown(typed.read(received("1706697000"))) shouldBe shown(utc("2024-01-31T10:30:00Z")) + slot(union("ts", "c_es", "UNION"), client.dateDialectVersion).read( + received("\"2024-01-31T10:30:00Z\"") + ) shouldBe utc("2024-01-31T10:30:00Z") + } + // a client that builds for no major (core's): no dialect, the dates of a format as stored + val core = forbidden(None) + core.dateDialectVersion shouldBe None + slot(union("ts", "ts", "UNION"), core.dateDialectVersion).sqlType shouldBe + MultiSearch.SetOperationType(SQLTypes.Varchar) + // the cluster's version where GET / answers + answering("8.18.3").dateDialectVersion shouldBe Some("8.18.3") + } + + it should "read them so through every wrapper of the client" in { + val base = forbidden(Some(8)) + val gateway: GatewayApi = new MetricsElasticClient(base, new MetricsCollector) + gateway.asInstanceOf[VersionApi].dateDialectVersion shouldBe Some("8") + base.calls.get() shouldBe 1 + } + + "Core's UNION ALL" should "read a converting date for a user denied GET /" in { + def rows(client: VersionDouble, sql: String, values: Seq[String]): Seq[(Any, Class[_])] = { + val (requests, _) = VersionDouble.resolved(sql) + val mappers = client.unionAllRowMappers(MultiSearch(requests)) + values.zipWithIndex.map { case (json, leg) => + shown(mappers(leg)(ListMap("k" -> received(json))).apply("k")) + } + } + val dates = Seq("\"2024-01-31\"", "\"2024/01/31\"") + val midnight = shown(utc("2024-01-31T00:00:00Z")) + rows(forbidden(), union("d", "c_c1"), dates) shouldBe Seq(midnight, midnight) + rows(answering("8.18.3"), union("d", "c_c1"), dates) shouldBe Seq(midnight, midnight) + // the statements that answered before answer the same values + Seq( + union("n", "x") -> Seq("5", "2.5"), + union("d", "d") -> Seq("\"2024-01-31\"", "\"2024-02-01\""), + union("ts", "ts") -> Seq("\"2024-01-31T10:30:00Z\"", "1706697000000"), + union("c_c1", "c_c1") -> Seq("\"2024/01/31\"", "\"2024/01/31\""), + union("tsz", "tsz") -> Seq("\"2024-01-31 10:30 UTC\"", "\"2024-01-31 10:30 CET\"") + ).foreach { case (sql, values) => + withClue(sql)(rows(forbidden(), sql, values) shouldBe rows(answering("8.18.3"), sql, values)) + } + // an agreeing date of a format of its own is read, a zone name kept as stored + rows(forbidden(), union("c_c1", "c_c1"), Seq("\"2024/01/31\"")) shouldBe Seq(midnight) + rows( + forbidden(), + union("tsz", "tsz"), + Seq("\"2024-01-31 10:30 UTC\"", "\"2024-01-31 10:30 CET\"") + ) shouldBe Seq( + shown(utc("2024-01-31T10:30:00Z")), + shown("2024-01-31 10:30 CET") + ) + } + + it should "carry the same reading to the legs of a one-shot _msearch" in { + val client = forbidden() + client.updateSchema("t", VersionDouble.schema) + client.search( + SelectStatement("SELECT d AS k FROM t LIMIT 5 UNION ALL SELECT c_c1 AS k FROM t LIMIT 5") + ) + val converters = client.legProjections.map(_.converters.map(_.apply(0))) + converters should have size 2 + converters.head.map(c => shown(c(received("\"2024-01-31\"")))) shouldBe + Some(shown(utc("2024-01-31T00:00:00Z"))) + converters(1).map(c => shown(c(received("\"2024/01/31\"")))) shouldBe + Some(shown(utc("2024-01-31T00:00:00Z"))) + } + + "GET /" should "be asked once per client, whatever the wrappers and the statements" in { + val appender = new ListAppender[ILoggingEvent]() + val logback = LoggerFactory.getLogger(VersionDouble.log.getName).asInstanceOf[LogbackLogger] + appender.start() + logback.addAppender(appender) + try { + val base = forbidden() + val gateway: GatewayApi = new MetricsElasticClient(base, new MetricsCollector) + val statements = Seq( + union("c_es", "ts", "UNION"), + union("d", "c_c1", "INTERSECT"), + union("tsz", "ts", "EXCEPT") + ) + (1 to 10).foreach { i => + val (requests, operators) = VersionDouble.resolved(statements(i % statements.size)) + SetOperationValues.legSchemas( + requests, + operators, + gateway.asInstanceOf[VersionApi].dateDialectVersion + ) shouldBe a[Right[_, _]] + } + (1 to 10).foreach { _ => + val (requests, _) = VersionDouble.resolved(union("d", "c_c1")) + base.unionAllRowMappers(MultiSearch(requests)) + } + base.calls.get() shouldBe 1 + appender.list.asScala.count(_.getLevel == Level.ERROR) shouldBe 1 + // a privilege granted later is seen after a restart: the client keeps its reading + base.answer = Some("8.18.3") + gateway.asInstanceOf[VersionApi].dateDialectVersion shouldBe Some("8") + base.dateDialectVersion shouldBe Some("8") + base.calls.get() shouldBe 1 + } finally logback.detachAppender(appender) + // a client whose GET / answers: once, the cluster's version + val answered = answering("8.18.3") + val wrapped: GatewayApi = new MetricsElasticClient(answered, new MetricsCollector) + (1 to 10).foreach { _ => + val (requests, operators) = VersionDouble.resolved(union("c_es", "ts", "UNION")) + SetOperationValues.legSchemas( + requests, + operators, + wrapped.asInstanceOf[VersionApi].dateDialectVersion + ) + answered.unionAllRowMappers(MultiSearch(VersionDouble.resolved(union("d", "c_c1"))._1)) + } + answered.calls.get() shouldBe 1 + wrapped.asInstanceOf[VersionApi].dateDialectVersion shouldBe Some("8.18.3") + } +} diff --git a/core/src/test/scala/app/softnetwork/elastic/client/DateRouteSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/DateRouteSpec.scala new file mode 100644 index 00000000..73c2cba9 --- /dev/null +++ b/core/src/test/scala/app/softnetwork/elastic/client/DateRouteSpec.scala @@ -0,0 +1,189 @@ +/* + * 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.`type`.SQLTypes +import app.softnetwork.elastic.sql.query.TemporalLiterals +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.time.{Clock, Instant, ZoneOffset} +import java.util.zip.GZIPInputStream +import scala.io.Source + +/** The route of a date position, decided from its format before any value ([[DateRoute.of]]), + * driven by the corpora MEASURED on Elasticsearch 6.8.23, 7.17.29 and 8.18.3 + * (`date-formats/es.tsv.gz`): a format is answered as stored EXACTLY when one of its values + * cannot be read by core for a reason the format alone gives (the row's 4th column: `jdk-offset`, + * `jdk-fraction`, `java6-name`, `jdk-ampm`, `zone-name`, `zone-id`); every other format -- the + * default included -- is typed, its values read by the format (a value-level reason, + * `json-double`, `erased-*`, `jdk-week`, fails at the value). + */ +class DateRouteSpec extends AnyFlatSpec with Matchers { + + /** The day the corpora were measured: Joda's two-digit years depend on it. */ + private val measuredOn = Clock.fixed(Instant.parse("2026-10-07T12:00:00Z"), ZoneOffset.UTC) + + /** The reasons the format alone gives. */ + private val formatCauses = + Set("jdk-offset", "jdk-fraction", "java6-name", "jdk-ampm", "zone-name", "zone-id") + + /** A row: the format (empty: none), the value as JSON, the instant indexed, and why core does not + * read it, if it does not. + */ + private def corpus(major: String): Seq[(String, String, Long, Option[String])] = { + val source = Source.fromInputStream( + new GZIPInputStream(getClass.getResourceAsStream(s"/date-formats/$major.tsv.gz")), + "UTF-8" + ) + try source + .getLines() + .filterNot(_.startsWith("#")) + .map { line => + line.split("\t", -1) match { + case Array(format, json, millis) => (format, json, millis.toLong, None) + case Array(format, json, millis, cause) => (format, json, millis.toLong, Some(cause)) + case other => throw new IllegalStateException(s"a corpus row: ${other.mkString("|")}") + } + } + .toList + finally source.close() + } + + private def dates(version: Option[String]): SetOperationValues.DateReading = + new SetOperationValues.DateReading(version, measuredOn) + + private def route(spec: String, version: String): DateRoute = + DateRoute.of(SQLTypes.Timestamp, Some(spec), dates(Some(version))) + + private def routesEveryFormat(major: String, version: String, formats: Int): Unit = { + val rows = corpus(major) + val byFormat = rows.groupBy(_._1) + byFormat.size shouldBe formats + val wrong = byFormat.toSeq.sortBy(_._1).flatMap { case (format, formatRows) => + val spec = if (format.isEmpty) TemporalLiterals.DefaultDateFormat else format + val causes = formatRows.flatMap(_._4).distinct.sorted + val expected = causes.exists(formatCauses) + route(spec, version) match { + case DateRoute.AsStored(reason) if expected => + if (reason.contains(s"'$spec'")) None + else Some(s"[$format] AsStored, its reason not naming the format: $reason") + case DateRoute.Typed if !expected => None + case other => + Some(s"[$format] ${causes.mkString("causes {", ", ", "}")} routed $other") + } + } + val asStored = byFormat.keys.count { format => + route(if (format.isEmpty) TemporalLiterals.DefaultDateFormat else format, version) + .isInstanceOf[DateRoute.AsStored] + } + info(s"$major on $version: ${byFormat.size} formats, $asStored as stored") + withClue(wrong.take(50).mkString("\n", "\n", "\n"))(wrong shouldBe empty) + } + + "Elasticsearch 8.18.3's date formats" should + "be answered as stored exactly when the format says a value is not read" in + routesEveryFormat("es8", "8.18.3", 4300) + + "Elasticsearch 7.17.29's date formats" should + "be answered as stored exactly when the format says a value is not read" in + routesEveryFormat("es7", "7.17.29", 4721) + + "Elasticsearch 6.8.23's date formats" should + "be answered as stored exactly when the format says a value is not read" in + routesEveryFormat("es6", "6.8.23", 6323) + + "A format" should "be answered as stored as a whole when ONE alternative is" in { + // a zone name after a name core reads + route("strict_date_optional_time||yyyy-MM-dd HH:mm z", "8.18.3") match { + case DateRoute.AsStored(reason) => + reason shouldBe + "core cannot read every value of the date format " + + "'strict_date_optional_time||yyyy-MM-dd HH:mm z' as Elasticsearch 8 does: its zone letter " + + "reads zone names, 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" + case other => fail(s"$other") + } + // a letter no measurement covers after an epoch + route("epoch_millis||yyyy-MM-dd B", "8.18.3") match { + case DateRoute.AsStored(reason) => + reason should include("'epoch_millis||yyyy-MM-dd B'") + reason should include("'yyyy-MM-dd B' is not a date format core reads") + case other => fail(s"$other") + } + // a region id + route("yyyy-MM-dd'T'HH:mm VV", "8.18.3") match { + case DateRoute.AsStored(reason) => reason should include("reads region ids") + case other => fail(s"$other") + } + // Joda-Time's zone name and region id + route("yyyy-MM-dd HH:mm z", "6.8.23") match { + case DateRoute.AsStored(reason) => reason should include("reads zone names") + case other => fail(s"$other") + } + route("yyyy-MM-dd HH:mm ZZZ", "6.8.23") match { + case DateRoute.AsStored(reason) => reason should include("reads region ids") + case other => fail(s"$other") + } + // 6.8's `8` mode: an AM/PM with no hour, known from the letters (quoted text aside) + route("8YYYY-ww a", "6.8.23") match { + case DateRoute.AsStored(reason) => + reason should include("reads an AM/PM with no hour") + case other => fail(s"$other") + } + route("8YYYY-ww hh a", "6.8.23") shouldBe DateRoute.Typed + route("8YYYY-ww 'h' a", "6.8.23") shouldBe a[DateRoute.AsStored] + // ...a reading of 6.8 only + route("YYYY-ww a", "7.17.29") shouldBe DateRoute.Typed + // an offset, `UTC` and the built-in names are read for sure + Seq("yyyy-MM-dd HH:mm XXX", "yyyy-MM-dd HH:mm Z", "epoch_second", "strict_date_time") + .foreach(spec => route(spec, "8.18.3") shouldBe DateRoute.Typed) + Seq("8.18.3", "7.17.29", "6.8.23").foreach { v => + route(TemporalLiterals.DefaultDateFormat, v) shouldBe DateRoute.Typed + } + } + + it should "be answered as stored, every one, when the major is not known" in { + DateRoute.of(SQLTypes.Timestamp, Some(TemporalLiterals.DefaultDateFormat), dates(None)) shouldBe + DateRoute.AsStored( + "the date format 'strict_date_optional_time||epoch_millis' is read as the Elasticsearch " + + "version does, which is not known" + ) + Seq("0.0.0", "5.6.16").foreach { v => + DateRoute.of(SQLTypes.Date, Some("epoch_second"), dates(Some(v))) shouldBe + DateRoute.AsStored( + s"the date format 'epoch_second' is read as the Elasticsearch version does, and core " + + s"reads dates of no $v" + ) + } + } + + "A position that is not a date read by its format" should "be typed, the version not asked" in { + var asked = 0 + def version: Option[String] = { asked += 1; Some("8.18.3") } + val reading = SetOperationValues.DateReading(version) + // a computed date, a number, a text: no format + DateRoute.of(SQLTypes.Timestamp, None, reading) shouldBe DateRoute.Typed + DateRoute.of(SQLTypes.Int, None, reading) shouldBe DateRoute.Typed + DateRoute.of(SQLTypes.Varchar, Some("yyyy-MM-dd HH:mm z"), reading) shouldBe DateRoute.Typed + asked shouldBe 0 + // a date read by its format: asked once for the statement + DateRoute.of(SQLTypes.Time, Some("HH:mm z"), reading) shouldBe a[DateRoute.AsStored] + DateRoute.of(SQLTypes.Date, Some("epoch_second"), reading) shouldBe DateRoute.Typed + asked shouldBe 1 + } +} diff --git a/core/src/test/scala/app/softnetwork/elastic/client/SetOperationLegSchemasSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/SetOperationLegSchemasSpec.scala new file mode 100644 index 00000000..7e5bab7b --- /dev/null +++ b/core/src/test/scala/app/softnetwork/elastic/client/SetOperationLegSchemasSpec.scala @@ -0,0 +1,702 @@ +/* + * 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.client.SetOperationValues.{LegSchema, LegSlot, Reading, Unreadable} +import app.softnetwork.elastic.sql.{StringValue, Value} +import app.softnetwork.elastic.sql.parser.Parser +import app.softnetwork.elastic.sql.query.{MultiSearch, SingleSearch, TemporalLiterals} +import app.softnetwork.elastic.sql.query.MultiSearch.{ + SetOperationColumn, + SetOperationType, + StructField +} +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`.{SQLType, SQLTypes} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.time.{Clock, Instant, LocalDate, ZoneOffset, ZonedDateTime} +import java.util.zip.GZIPInputStream +import scala.collection.immutable.ListMap +import scala.io.Source +import scala.util.{Failure, Success, Try} + +/** Each leg of a set operation TYPED BEFORE ANY VALUE ([[SetOperationValues.legSchemas]]): per + * column, the leg's type at every depth and the reader of its values -- a date whose format core + * cannot read for sure `VARCHAR`, the text as stored, every row, whatever their order; every other + * date read as [[SetOperationValues.legReaders]] reads it. Every value goes the way core receives + * it: its JSON, read by the response parser (`ElasticConversion.jsonNodeToAny`). + */ +class SetOperationLegSchemasSpec extends AnyFlatSpec with Matchers { + + private def format(f: String): ListMap[String, Value[_]] = + ListMap[String, Value[_]]("format" -> StringValue(f)) + + private val zoneName = "yyyy-MM-dd HH:mm z" + + 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_def", SQLTypes.Timestamp), + Column("tsz", SQLTypes.Timestamp, options = format(zoneName)), + Column("ts", SQLTypes.Timestamp), + Column("d", SQLTypes.Date), + Column("n", SQLTypes.Int), + Column("x", SQLTypes.Double), + Column("f", SQLTypes.Real), + Column("s", SQLTypes.Keyword), + Column("bin", SQLTypes.VarBinary), + Column("meta", SQLTypes.Any), + Column("o", SQLTypes.Struct, multiFields = List(Column("meta", SQLTypes.Any))), + Column("fl", SQLTypes.Struct), + Column( + "oz", + SQLTypes.Struct, + multiFields = List(Column("z", SQLTypes.Timestamp, options = format(zoneName))) + ), + Column("ozs", SQLTypes.Struct, multiFields = List(Column("z", SQLTypes.Keyword))), + Column("od", SQLTypes.Struct, multiFields = List(Column("t", SQLTypes.Timestamp))), + Column( + "obj", + SQLTypes.Struct, + multiFields = List( + Column("n", SQLTypes.Int), + Column("e", SQLTypes.Timestamp, options = format("epoch_second")) + ) + ), + Column("ofr", SQLTypes.Struct, multiFields = List(Column("f", SQLTypes.Real))), + Column("o2", SQLTypes.Struct, multiFields = List(Column("n", SQLTypes.Int))), + Column( + "p", + SQLTypes.Struct, + multiFields = List( + Column( + "m", + SQLTypes.Struct, + multiFields = List(Column("e", SQLTypes.Timestamp, options = format("epoch_second"))) + ) + ) + ), + Column("q", SQLTypes.Struct, multiFields = List(Column("m", SQLTypes.Any))), + Column("r", SQLTypes.Struct, multiFields = List(Column("m", SQLTypes.Struct))), + Column("az", SQLTypes.Array(SQLTypes.Timestamp), options = format(zoneName)), + Column("ae", SQLTypes.Array(SQLTypes.Timestamp), options = format("epoch_second")), + Column("ats", SQLTypes.Array(SQLTypes.Timestamp)), + Column("ar", SQLTypes.Array(SQLTypes.Real)) + ) + ) + + private val es8 = Some("8.18.3") + + private val varchar = SetOperationType(SQLTypes.Varchar) + + 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 schemas(sql: String, version: => Option[String] = es8): Seq[Option[LegSchema]] = { + val (requests, operators) = resolved(sql) + SetOperationValues.legSchemas(requests, operators, version) match { + case Right(legs) => legs + case Left(refusal) => fail(s"[$sql] refused: $refusal") + } + } + + /** Leg `leg`'s slot at `position`. */ + private def slot( + sql: String, + leg: Int = 0, + position: Int = 0, + version: => Option[String] = es8 + ): LegSlot = + schemas(sql, version)(leg).flatMap(_.slots(position)) match { + case Some(s) => s + case None => fail(s"[$sql] leg $leg has no slot at $position") + } + + /** The slots of every leg: `None` for a `*` leg, else each position's slot or `None`. */ + private def slots(sql: String): Seq[Option[Seq[Option[LegSlot]]]] = + schemas(sql).map(_.map(_.slots)) + + /** The reader [[SetOperationValues.legReaders]] gives leg `leg` at `position` -- `null` where it + * reads nothing. + */ + private def legReader(sql: String, leg: Int = 0, position: Int = 0): Any => Any = { + val (requests, operators) = resolved(sql) + SetOperationValues.legReaders(requests, operators, es8) match { + case Right(legs) => legs(leg).map(_(position)).orNull + 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) + + private def union(a: String, b: String, operator: String = "UNION"): String = + s"SELECT $a AS k FROM t $operator SELECT $b AS k FROM t" + + /** A value and its class. */ + private def shown(v: Any): (Any, Class[_]) = + v match { + case null => (null, classOf[Null]) + case other => (other, other.getClass) + } + + private def utc(s: String): ZonedDateTime = ZonedDateTime.parse(s) + + private val at1030 = utc("2024-01-31T10:30:00Z") + + // --------------------------------------------------------------------------------------------- + // Order-independent VARCHAR + // --------------------------------------------------------------------------------------------- + + "A zone-name date" should "be VARCHAR before any value, every row as stored, in any order" in { + Seq("8.18.3", "7.17.29", "6.8.23").foreach { v => + var asked = 0 + def version: Option[String] = { asked += 1; Some(v) } + val legs = schemas(union("tsz", "ts"), version) + // decided with no value read: the version asked once + asked shouldBe 1 + val tsz = legs.head.flatMap(_.slots.head).getOrElse(fail(s"$v: no slot")) + tsz.sqlType shouldBe varchar + tsz.asStored.map(_.path) shouldBe Seq("tsz") + tsz.asStored.head.reason should include(s"'$zoneName'") + tsz.asStored.head.reason should include("zone names") + legs(1).flatMap(_.slots.head).map(_.sqlType) shouldBe + Some(SetOperationType(SQLTypes.Timestamp, format = Some(TemporalLiterals.DefaultDateFormat))) + val utcText = "2024-01-31 10:30 UTC" + val cetText = "2024-01-31 10:30 CET" + val rows = Seq.fill(1000)(utcText) :+ cetText + Seq(rows, rows.reverse, Seq.fill(10)(cetText), Seq.fill(10)(utcText)).foreach { texts => + val read = texts.map(text => tsz.read(received(s"\"$text\""))) + withClue(s"$v: ")(read shouldBe texts) + read.map(_.getClass).distinct shouldBe Seq(classOf[String]) + } + asked shouldBe 1 + } + } + + // --------------------------------------------------------------------------------------------- + // Typed readers are legReaders'; the last release's forms + // --------------------------------------------------------------------------------------------- + + "A date read by its format" should "be its declared type, read as legReaders reads it" in { + Seq( + ("c_es", "c_es", "1706697000"), + ("c_es", "ts", "1706697000"), + ("c_em", "c_em", "1706697000000"), + ("c_def", "ts", "1706697000000") + ).foreach { case (a, b, json) => + val sql = union(a, b) + val s = slot(sql) + s.sqlType.sqlType shouldBe SQLTypes.Timestamp + s.asStored shouldBe empty + shown(s.read(received(json))) shouldBe shown(at1030) + shown(s.read(received(json))) shouldBe shown(legReader(sql)(received(json))) + } + // a zone name stays VARCHAR, agreeing or converting + Seq("tsz", "s", "ts").foreach(b => slot(union("tsz", b)).sqlType shouldBe varchar) + slot(union("tsz", "s")).read(received("\"2024-01-31 10:30 UTC\"")) shouldBe + "2024-01-31 10:30 UTC" + } + + // --------------------------------------------------------------------------------------------- + // Every corpus row + // --------------------------------------------------------------------------------------------- + + /** The day the corpora were measured: Joda's two-digit years depend on it. */ + private val measuredOn = Clock.fixed(Instant.parse("2026-10-07T12:00:00Z"), ZoneOffset.UTC) + + private val valueCauses = Set("json-double", "erased-offset", "erased-spelling", "jdk-week") + + private def corpus(major: String): Seq[(String, String, Long, Option[String])] = { + val source = Source.fromInputStream( + new GZIPInputStream(getClass.getResourceAsStream(s"/date-formats/$major.tsv.gz")), + "UTF-8" + ) + try source + .getLines() + .filterNot(_.startsWith("#")) + .map { line => + line.split("\t", -1) match { + case Array(f, json, millis) => (f, json, millis.toLong, None) + case Array(f, json, millis, cause) => (f, json, millis.toLong, Some(cause)) + case other => throw new IllegalStateException(s"a corpus row: ${other.mkString("|")}") + } + } + .toList + finally source.close() + } + + private def readsEveryRow(major: String, version: String, size: Int): Unit = { + val rows = corpus(major) + rows.size shouldBe size + val dates = new SetOperationValues.DateReading(Some(version), measuredOn) + // the text leg's reader: a KEYWORD beside a KEYWORD + val keyword = SetOperationValues.converter(SQLTypes.Varchar, SQLTypes.Varchar) + var typedRead, typedFailed, storedRead = 0 + val wrong = rows.groupBy(_._1).toSeq.sortBy(_._1).flatMap { case (f, formatRows) => + val spec = if (f.isEmpty) TemporalLiterals.DefaultDateFormat else f + val declared = SetOperationType(SQLTypes.Timestamp, format = Some(spec)) + val column = + SetOperationColumn( + SetOperationType(SQLTypes.Timestamp), + Seq(Some(declared), Some(declared)) + ) + val legSlot = SetOperationValues.legSlot(column, 0, "k", "v", dates) + // legReaders' reader of the same column (`legReader`'s own call) + val reference = SetOperationValues.positionConverter( + column.result, + column.branches, + 0, + "k", + "v", + dates, + Reading.Declared(Unreadable.Fail), + agrees = Some(!column.converts) + ) + (DateRoute.of(SQLTypes.Timestamp, Some(spec), dates), legSlot) match { + case (_, None) => Seq(s"[$f] no slot") + case (DateRoute.Typed, Some(s)) => + (if (s.sqlType != declared || s.asStored.nonEmpty) Seq(s"[$f] typed as $s") else Nil) ++ + formatRows.flatMap { case (_, json, millis, cause) => + val value = received(json) + (cause, Try(s.read(value)), Try(reference(value))) match { + case (None, Success(z: ZonedDateTime), Success(r)) + if shown(z) == shown(r) && z.toInstant.toEpochMilli == millis => + typedRead += 1 + None + case (Some(c), Failure(e: IllegalArgumentException), Failure(r)) + if valueCauses(c) && e.getMessage == r.getMessage && + e.getMessage.contains("field v") && e.getMessage.contains(s"'$spec'") => + typedFailed += 1 + None + case (c, got, want) => Some(s"[$f] $json ($c): read $got, legReaders $want") + } + } + case (DateRoute.AsStored(reason), Some(s)) => + (if (s.sqlType != varchar || s.asStored != Seq(LegSlot.AsStored("v", reason))) + Seq(s"[$f] as stored, typed as $s") + else Nil) ++ + formatRows.flatMap { case (_, json, _, cause) => + val value = received(json) + (Try(s.read(value)), Try(keyword(value))) match { + case (Success(text: String), Success(k)) if text == k => + storedRead += 1 + None + case (got, want) => Some(s"[$f] $json ($cause): read $got, the text leg $want") + } + } + } + } + info( + s"$major on $version: ${rows.size} rows -- $typedRead read by their format, $typedFailed " + + s"failing at the value, $storedRead as stored" + ) + withClue(wrong.take(50).mkString("\n", "\n", "\n"))(wrong shouldBe empty) + (typedRead + typedFailed + storedRead) shouldBe size + } + + "Every value of Elasticsearch 8.18.3's corpus" should "be read by its slot's reader" in + readsEveryRow("es8", "8.18.3", 36109) + + "Every value of Elasticsearch 7.17.29's corpus" should "be read by its slot's reader" in + readsEveryRow("es7", "7.17.29", 47065) + + "Every value of Elasticsearch 6.8.23's corpus" should "be read by its slot's reader" in + readsEveryRow("es6", "6.8.23", 51515) + + // --------------------------------------------------------------------------------------------- + // Positions and depth + // --------------------------------------------------------------------------------------------- + + "A date answered as stored" should "be VARCHAR at every position and depth" in { + val cet = "2024-01-31 10:30 CET" + // agreeing, converting to VARCHAR and to TIMESTAMP, multi-step + Seq(union("tsz", "tsz"), union("tsz", "s"), union("tsz", "d"), union("tsz", "ts")).foreach { + sql => + val s = slot(sql) + s.sqlType shouldBe varchar + s.asStored.map(_.path) shouldBe Seq("tsz") + s.read(received(s"\"$cet\"")) shouldBe cet + } + slot(union("tsz", "tsz"), 1).sqlType shouldBe varchar + slot(union("tsz", "d"), 1).sqlType.sqlType shouldBe SQLTypes.Date + val multiStep = + "SELECT tsz AS k FROM t UNION SELECT ts AS k FROM t INTERSECT SELECT tsz AS k FROM t" + Seq(0, 2).foreach(leg => slot(multiStep, leg).sqlType shouldBe varchar) + slot(multiStep, 1).sqlType.sqlType shouldBe SQLTypes.Timestamp + // a struct's field: STRUCT(z VARCHAR), its path `oz.z` + Seq(union("oz", "oz"), union("oz", "ozs")).foreach { sql => + val s = slot(sql) + s.sqlType shouldBe SetOperationType(SQLTypes.Struct, Seq(StructField("z", varchar))) + s.asStored.map(_.path) shouldBe Seq("oz.z") + s.read(received(s"""{"z":"$cet"}""")) shouldBe ListMap("z" -> cet) + } + // a list's element: ARRAY + val az = slot(union("az", "az")) + az.sqlType shouldBe + SetOperationType(SQLTypes.Array(SQLTypes.Varchar), element = Some(varchar)) + az.asStored.map(_.path) shouldBe Seq("az") + az.read(received(s"""["2024-01-31 10:30 UTC","$cet"]""")) shouldBe + List("2024-01-31 10:30 UTC", cet) + az.read(received(s"\"$cet\"")) shouldBe List(cet) + } + + "A date read by its format" should "keep its type at every position and depth" in { + // (statement, leg, JSON): each read as legReaders reads it + val cases = Seq( + (union("c_es", "c_es"), 0, "1706697000"), + (union("c_es", "s"), 0, "1706697000"), + (union("c_es", "d"), 0, "1706697000"), + (union("ts", "ts"), 0, "\"2024-01-31T10:30:00Z\""), + (union("ts", "s"), 0, "\"2024-01-31T10:30:00Z\""), + (union("ts", "d"), 1, "\"2024-01-31\""), + ( + "SELECT c_es AS k FROM t UNION SELECT ts AS k FROM t INTERSECT SELECT c_es AS k FROM t", + 2, + "1706697000" + ), + (union("obj", "obj"), 0, """{"n":1,"e":1706697000}"""), + (union("od", "od"), 0, """{"t":"2024-01-31T10:30:00Z"}"""), + (union("ae", "ae"), 0, "[1706697000,1706697001]"), + (union("ats", "ats"), 0, "[\"2024-01-31T10:30:00Z\"]") + ) + val wrong = cases.flatMap { case (sql, leg, json) => + val s = slot(sql, leg) + val got = Try(s.read(received(json))).map(shown) + val want = Try(legReader(sql, leg)(received(json))).map(shown) + if (s.asStored.isEmpty && got.isSuccess && got == want) None + else Some(s"[$sql] leg $leg <- $json: $s read $got, legReaders $want") + } + wrong shouldBe empty + slot(union("obj", "obj")).read(received("""{"n":1,"e":1706697000}""")) shouldBe + ListMap("e" -> at1030, "n" -> 1) + slot(union("obj", "obj")).sqlType.fields.map(f => f.name -> f.valueType) shouldBe Seq( + "e" -> SetOperationType(SQLTypes.Timestamp, format = Some("epoch_second")), + "n" -> SetOperationType(SQLTypes.Int) + ) + slot(union("ae", "ae")).sqlType.element shouldBe + Some(SetOperationType(SQLTypes.Timestamp, format = Some("epoch_second"))) + } + + // --------------------------------------------------------------------------------------------- + // Several values + // --------------------------------------------------------------------------------------------- + + "A value of several values" should "fail in core's words, typed or as stored" in { + def failure(read: Any => Any, json: String): String = + intercept[IllegalArgumentException](read(received(json))).getMessage + val typed = slot(union("c_es", "c_es")).read + val stored = slot(union("tsz", "tsz")).read + Seq("""["a","b"]""", "[[1706697000,1706697001]]").foreach { json => + failure(typed, json) shouldBe failure(legReader(union("c_es", "c_es")), json) + failure(typed, json) shouldBe + "Conversion Error: Unimplemented type for cast (TIMESTAMP -> TIMESTAMP[]) when casting " + + "from source column k" + failure(stored, json) shouldBe + "Conversion Error: Unimplemented type for cast (TIMESTAMP -> VARCHAR[]) when casting from " + + "source column k" + } + // Elasticsearch's one-element array is ONE value + stored(received("""["c"]""")) shouldBe "c" + } + + // --------------------------------------------------------------------------------------------- + // Value-typed positions + // --------------------------------------------------------------------------------------------- + + "A position typed as its values come" should "have no slot where legReaders reads nothing" in { + // a column whose type is not known: a WITH aliasing inside its CTE, every derived table, in + // every leg; a column of type ANY; a NULL literal + val anyColumns = Seq( + "WITH a AS (SELECT tsz AS k FROM t) SELECT k FROM a UNION SELECT ts AS k FROM t", + "SELECT k FROM (SELECT tsz AS k FROM t) dd UNION SELECT ts AS k FROM t", + "SELECT dd.k FROM (SELECT tsz AS k FROM t) dd UNION SELECT ts AS k FROM t", + "SELECT ts AS k FROM t UNION SELECT k FROM (SELECT tsz AS k FROM t) dd", + union("meta", "meta") + ) + anyColumns.foreach { sql => + val (requests, operators) = resolved(sql) + MultiSearch.columnTypes(requests, operators).map(_.map(_.resultType)) shouldBe + Right(Seq(SQLTypes.Any)) + slots(sql) shouldBe Seq(Some(Seq(None)), Some(Seq(None))) + Seq(0, 1).foreach(leg => legReader(sql, leg) shouldBe null) + } + slots(union("n", "NULL")).map(_.map(_.map(_.map(_.sqlType)))) shouldBe + Seq(Some(Seq(Some(SetOperationType(SQLTypes.Int)))), Some(Seq(None))) + legReader(union("n", "NULL"), 1) shouldBe null + // a SELECT * leg: no schema, whatever the width of the others + slots("SELECT n FROM t UNION ALL SELECT * FROM t").map(_.isDefined) shouldBe Seq(true, false) + slots("SELECT n, x FROM t UNION ALL SELECT * FROM t").map(_.isDefined) shouldBe + Seq(true, false) + } + + it should "keep the plain column's slot in a WITH that passes it through" in { + val passThrough = + "WITH a AS (SELECT tsz FROM t) SELECT tsz AS k FROM a UNION SELECT ts AS k FROM t" + slots(passThrough).map(_.map(_.map(_.map(s => (s.sqlType, s.asStored))))) shouldBe + slots(union("tsz", "ts")).map(_.map(_.map(_.map(s => (s.sqlType, s.asStored))))) + slot(passThrough).sqlType shouldBe varchar + val epochs = + "WITH a AS (SELECT c_es FROM t) SELECT c_es AS k FROM a UNION SELECT ts AS k FROM t" + shown(slot(epochs).read(received("1706697000"))) shouldBe shown(at1030) + } + + it should "have no slot at a type not known at any depth" in { + // STRUCT(meta ANY), a STRUCT whose fields are not known: legReaders reads them, arrow types + // their values as they come + Seq(union("o", "o"), union("fl", "fl")).foreach { sql => + slots(sql) shouldBe Seq(Some(Seq(None)), Some(Seq(None))) + } + } + + it should "have no slot where ANOTHER leg leaves the column's type not known at some depth" in { + var asked = 0 + def version: Option[String] = { asked += 1; es8 } + // the column is a STRUCT whose fields are not known (`fl`), or holds one (`r.m`) or an `ANY` + // (`q.m`) where this leg declares its fields: legReader reads the leg's value there as it + // comes, not as the leg's type -- no slot for any leg, and no version asked for `p`'s date + Seq( + union("obj", "fl"), + union("fl", "obj"), + union("obj", "fl", "UNION ALL"), + union("oz", "fl"), + union("p", "q"), + union("p", "r") + ).foreach { sql => + withClue(s"[$sql] ") { + schemas(sql, version).map(_.map(_.slots)) shouldBe Seq(Some(Seq(None)), Some(Seq(None))) + asked shouldBe 0 + } + } + // the column's type known at every depth: every leg keeps its slot + Seq(union("obj", "o2"), union("obj", "obj"), union("oz", "oz"), union("p", "p")).foreach { + sql => + withClue(s"[$sql] ")( + slots(sql).map(_.map(_.map(_.isDefined))) shouldBe Seq(Some(Seq(true)), Some(Seq(true))) + ) + } + slot(union("p", "p")).read(received("""{"m":{"e":1706697000}}""")) shouldBe + ListMap("m" -> ListMap("e" -> at1030)) + } + + /** One value per column of [[schema]], its JSON as Elasticsearch stores it -- every object + * holding, at every depth, a key its mapping does not declare (`zz`), which a reader typing the + * object leaves out. + */ + private val stored: ListMap[String, String] = ListMap( + "c_es" -> "1706697000", + "c_em" -> "1706697000000", + "c_c1" -> "\"2024/01/31\"", + "c_def" -> "\"2024-01-31T10:30:00Z\"", + "tsz" -> "\"2024-01-31 10:30 CET\"", + "ts" -> "\"2024-01-31T10:30:00Z\"", + "d" -> "\"2024-01-31\"", + "n" -> "1", + "x" -> "1.5", + "f" -> "1.5", + "s" -> "\"a\"", + "bin" -> "\"aGk=\"", + "meta" -> """{"a":1}""", + "o" -> """{"meta":{"a":1},"zz":1}""", + "fl" -> """{"a":1,"zz":1}""", + "oz" -> """{"z":"2024-01-31 10:30 CET","zz":1}""", + "ozs" -> """{"z":"a","zz":1}""", + "od" -> """{"t":"2024-01-31T10:30:00Z","zz":1}""", + "obj" -> """{"n":1,"e":1706697000,"zz":1}""", + "ofr" -> """{"f":1.5,"zz":1}""", + "o2" -> """{"n":1,"zz":1}""", + "p" -> """{"m":{"e":1706697000,"zz":1},"zz":1}""", + "q" -> """{"m":{"a":1},"zz":1}""", + "r" -> """{"m":{"a":1},"zz":1}""", + "az" -> """["2024-01-31 10:30 UTC","2024-01-31 10:30 CET"]""", + "ae" -> "[1706697000,1706697001]", + "ats" -> """["2024-01-31T10:30:00Z","2024-01-31T11:30:00Z"]""", + "ar" -> "[1.5,2.5]" + ) + + /** The class of each scalar type a slot of [[schema]] holds. */ + private val scalarClass: Map[SQLType, Class[_]] = Map( + SQLTypes.Varchar -> classOf[String], + SQLTypes.Int -> classOf[java.lang.Integer], + SQLTypes.Double -> classOf[java.lang.Double], + SQLTypes.Date -> classOf[LocalDate], + SQLTypes.Timestamp -> classOf[ZonedDateTime] + ) + + /** Whether `v` is of `t`'s class at every depth: a `STRUCT` a `ListMap` of EXACTLY its fields, in + * its order; a list a `List` of its element's class; `null` of any type. + */ + private def conforms(t: SetOperationType, v: Any): Boolean = + v == null || (t.sqlType match { + case SQLTypes.Struct => + v match { + case m: ListMap[_, _] => + val fields = m.asInstanceOf[ListMap[String, Any]] + fields.keys.toSeq == t.fields.map(_.name) && + t.fields.forall(f => conforms(f.valueType, fields(f.name))) + case _ => false + } + case SQLTypes.Array(_) => + v match { + case l: List[_] => t.element.exists(e => l.forall(conforms(e, _))) + case _ => false + } + case scalar => scalarClass.get(scalar).exists(_.isInstance(v)) + }) + + "Every slot" should "answer its type's class at every depth, over every pair of columns" in { + val names = stored.keys.toSeq + names.toSet shouldBe schema.columns.map(_.name).toSet + val mismatches = Seq.newBuilder[String] + var statements = 0 + var refused = 0 + var slotsRead = 0 + for (a <- names; b <- names; operator <- Seq("UNION", "UNION ALL")) { + val sql = union(a, b, operator) + val (requests, operators) = resolved(sql) + statements += 1 + SetOperationValues.legSchemas(requests, operators, es8) match { + case Left(_) => refused += 1 + case Right(legs) => + legs.zip(Seq(a, b)).foreach { case (leg, column) => + leg.flatMap(_.slots.head).foreach { s => + slotsRead += 1 + Try(s.read(received(stored(column)))) match { + case Success(v) if conforms(s.sqlType, v) => + case outcome => mismatches += s"[$sql] $column ${s.sqlType}: ${outcome.map(shown)}" + } + } + } + } + } + mismatches.result() shouldBe empty + // the population: 28 x 28 columns x 2 operators, 992 statements refused (`branchTypes`); 544 + // slots read -- 640 before no slot was given where ANOTHER leg leaves the column's type not + // known inside, 32 of which answered another class than their type's + (statements, refused, slotsRead) shouldBe ((1568, 992, 544)) + } + + it should "have no slot at NUMERIC or TEMPORAL, the other legs keeping theirs" in { + Seq("EXTRACT(YEAR FROM ts)", "DATEDIFF(ts, d)").foreach { computed => + val legs = slots(union(computed, "x")) + legs.head shouldBe Some(Seq(None)) + legs(1).flatMap(_.head).map(_.sqlType) shouldBe Some(SetOperationType(SQLTypes.Double)) + } + // a constructed TEMPORAL leg (no SQL form yields one) + val temporal = SetOperationColumn( + SetOperationType(SQLTypes.Temporal), + Seq( + Some(SetOperationType(SQLTypes.Temporal)), + Some(SetOperationType(SQLTypes.Timestamp, format = Some("epoch_second"))) + ) + ) + val dates = SetOperationValues.DateReading(es8) + SetOperationValues.legSlot(temporal, 0, "k", "", dates) shouldBe None + SetOperationValues.legSlot(temporal, 1, "k", "c_es", dates).map(_.sqlType.sqlType) shouldBe + Some(SQLTypes.Timestamp) + // DATE_TRUNC is a TIMESTAMP, computed: no format, read as legReaders reads it + val truncated = union("DATE_TRUNC(ts, DAY)", "ts") + slot(truncated).sqlType shouldBe SetOperationType(SQLTypes.Timestamp) + shown(slot(truncated).read(received("\"2024-01-31T00:00:00Z\""))) shouldBe + shown(legReader(truncated)(received("\"2024-01-31T00:00:00Z\""))) + } + + it should "have no slot at VARBINARY or REAL at any depth, the other legs keeping theirs" in { + Seq(union("bin", "bin"), union("f", "f"), union("ofr", "ofr"), union("ar", "ar")).foreach { + sql => slots(sql) shouldBe Seq(Some(Seq(None)), Some(Seq(None))) + } + val text = slots(union("s", "bin")) + text.head.flatMap(_.head).map(_.sqlType) shouldBe Some(varchar) + text(1) shouldBe Some(Seq(None)) + val double = slots(union("x", "f")) + double.head.flatMap(_.head).map(_.sqlType) shouldBe Some(SetOperationType(SQLTypes.Double)) + double(1) shouldBe Some(Seq(None)) + } + + // --------------------------------------------------------------------------------------------- + // The version + // --------------------------------------------------------------------------------------------- + + "A leg's schema" should "read its dates as the major the statement runs on" in { + val sql = union("c_es", "c_es") + slot(sql, version = Some("8.18.3")).read(received("\"1706697000.5\"")) shouldBe + utc("2024-01-31T10:30:00.500Z") + slot(sql, version = Some("7.17.29")).read(received("\"1706697000.5\"")) shouldBe + utc("2024-01-31T10:30:00.500Z") + slot(sql, version = Some("6.8.23")).read(received("\"1706697000.5\"")) shouldBe at1030 + } + + it should "answer every date of a format as stored when the major is not known" in { + val sql = + "SELECT c_es AS a, ts AS b, tsz AS c, DATE_TRUNC(ts, DAY) AS e FROM t UNION " + + "SELECT c_es AS a, ts AS b, tsz AS c, DATE_TRUNC(ts, DAY) AS e FROM t" + Seq(None -> "which is not known", Some("0.0.0") -> "no 0.0.0", Some("5.6.16") -> "no 5.6.16") + .foreach { case (version, named) => + val leg = schemas(sql, version).head.getOrElse(fail("no schema")).slots + leg.take(3).map(_.map(_.sqlType)) shouldBe Seq.fill(3)(Some(varchar)) + leg.take(3).flatMap(_.toSeq.flatMap(_.asStored)).map(_.path) shouldBe + Seq("c_es", "ts", "tsz") + leg.take(3).flatMap(_.toSeq.flatMap(_.asStored)).foreach(_.reason should include(named)) + leg.head.map(_.read(received("1706697000"))) shouldBe Some("1706697000") + // a computed date: typed + leg(3).map(_.sqlType) shouldBe Some(SetOperationType(SQLTypes.Timestamp)) + leg(3).map(s => shown(s.read(received("\"2024-01-31T00:00:00Z\"")))) shouldBe + Some(shown(utc("2024-01-31T00:00:00Z"))) + } + } + + it should "ask the version at most once, never for a statement with no formatted date slot" in { + var asked = 0 + def version: Option[String] = { asked += 1; es8 } + schemas("SELECT c_es, ts, n FROM t UNION ALL SELECT c_c1, d, x FROM t", version) + asked shouldBe 1 + schemas(union("n", "x"), version) + schemas(union("s", "s"), version) + schemas(union("DATE_TRUNC(ts, DAY)", "DATE_TRUNC(d, MONTH)"), version) + asked shouldBe 1 + // a date read by its format where no slot is: a column whose type is not known (`ANY`) -- a + // WITH aliasing inside its CTE, a derived table, in any leg -- which legReaders reads nothing at + Seq( + "SELECT k FROM (SELECT tsz AS k FROM t) dd UNION SELECT ts AS k FROM t", + "SELECT dd.k FROM (SELECT tsz AS k FROM t) dd UNION SELECT ts AS k FROM t", + "WITH a AS (SELECT tsz AS k FROM t) SELECT k FROM a UNION SELECT ts AS k FROM t", + "SELECT ts AS k FROM t UNION SELECT k FROM (SELECT tsz AS k FROM t) dd", + "SELECT k FROM (SELECT c_c1 AS k FROM t) dd UNION SELECT c_c1 AS k FROM t" + ).foreach { sql => + val before = asked + schemas(sql, version).map(_.map(_.slots)) shouldBe Seq(Some(Seq(None)), Some(Seq(None))) + withClue(s"[$sql] ")(asked shouldBe before) + } + } + + it should "answer a refused statement with its refusal, word for word" in { + val (refused, operators) = resolved(union("d", "n")) + SetOperationValues.legSchemas(refused, operators, es8) shouldBe + MultiSearch.branchTypes(refused, operators).map(_ => Nil) + } +} diff --git a/documentation/sql/dql_statements.md b/documentation/sql/dql_statements.md index 20a507cc..c5fca3fb 100644 --- a/documentation/sql/dql_statements.md +++ b/documentation/sql/dql_statements.md @@ -780,6 +780,11 @@ SELECT ts AS k FROM t UNION ALL SELECT s AS k FROM t; The readings were measured on Elasticsearch 6.8.23, 7.17.29 and 8.18.3. The other minor versions of a major are read as its measured one, and Elasticsearch 9 as Elasticsearch 8. +The major is the cluster's version, asked once per client (`GET /`). For a user whose role may not +read it, dates are read as the major the client is built for (the Elasticsearch 8 client reads them +as Elasticsearch 8), at a position that converts as where the branches agree. The version is kept +for the life of the client: a privilege granted later is seen once the client restarts. + A value with several values — a multi-valued field — has no one value of its position's type, so at a position that converts its values it fails the statement, in the words DuckDB uses for the same failure: `Conversion Error: Unimplemented type for cast (INTEGER -> VARCHAR[]) when casting diff --git a/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestVersionApi.scala b/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestVersionApi.scala index aa4b2181..d0120193 100644 --- a/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestVersionApi.scala +++ b/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestVersionApi.scala @@ -48,6 +48,10 @@ trait JestVersionApi extends VersionApi with JestClientHelpers { case _ => false } + override private[client] def buildElasticMajor: Option[Int] = Some( + JestClientBuildInfo.elasticMajor + ) + override private[client] def executeVersion(): result.ElasticResult[String] = executeJestAction( "version", diff --git a/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala b/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala index 6789fcbf..e02a4949 100644 --- a/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala +++ b/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala @@ -159,6 +159,9 @@ trait RestHighLevelClientApi */ trait RestHighLevelClientVersionApi extends VersionApi with RestHighLevelClientHelpers { _: RestHighLevelClientCompanion => + override private[client] def buildElasticMajor: Option[Int] = Some( + RestClientBuildInfo.elasticMajor + ) override private[client] def executeVersion(): ElasticResult[String] = executeRestLowLevelAction[String]( operation = "version", diff --git a/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala b/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala index 7e9b7ab3..1e13330d 100644 --- a/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala +++ b/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala @@ -231,6 +231,10 @@ trait RestHighLevelClientApi trait RestHighLevelClientVersionApi extends VersionApi with RestHighLevelClientHelpers { _: RestHighLevelClientCompanion => + override private[client] def buildElasticMajor: Option[Int] = Some( + RestClientBuildInfo.elasticMajor + ) + override private[client] def executeVersion(): ElasticResult[String] = executeRestLowLevelAction[String]( operation = "version", diff --git a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala index befb890f..87555f3f 100644 --- a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala +++ b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala @@ -162,6 +162,9 @@ trait JavaClientApi */ trait JavaClientVersionApi extends VersionApi with JavaClientHelpers { _: JavaClientCompanion => + override private[client] def buildElasticMajor: Option[Int] = Some( + JavaClientBuildInfo.elasticMajor + ) override private[client] def executeVersion(): ElasticResult[String] = executeJavaAction( operation = "version", diff --git a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala index 30c96e07..9ee8e1b8 100644 --- a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala +++ b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala @@ -161,6 +161,9 @@ trait JavaClientApi */ trait JavaClientVersionApi extends VersionApi with JavaClientHelpers { _: JavaClientCompanion => + override private[client] def buildElasticMajor: Option[Int] = Some( + JavaClientBuildInfo.elasticMajor + ) override private[client] def executeVersion(): ElasticResult[String] = executeJavaAction( operation = "version",