diff --git a/bridge/src/main/scala/app/softnetwork/elastic/sql/bridge/ElasticAggregation.scala b/bridge/src/main/scala/app/softnetwork/elastic/sql/bridge/ElasticAggregation.scala index c8e51957..63770b39 100644 --- a/bridge/src/main/scala/app/softnetwork/elastic/sql/bridge/ElasticAggregation.scala +++ b/bridge/src/main/scala/app/softnetwork/elastic/sql/bridge/ElasticAggregation.scala @@ -17,7 +17,6 @@ package app.softnetwork.elastic.sql.bridge import app.softnetwork.elastic.sql.{PainlessContext, PainlessContextType, PainlessTarget} -import app.softnetwork.elastic.sql.`type`.SQLTemporal import app.softnetwork.elastic.sql.query.{ Asc, BucketIncludesExcludes, @@ -503,8 +502,11 @@ object ElasticAggregation { } var agg: Aggregation = { + // Story IDENT-4 -- ONE derivation of "this level is a date histogram", shared with the + // HAVING classifier (`SingleSearch.dateHistogramLevelOf`), which refuses a key filter + // this level could not carry. bucket.out match { - case _: SQLTemporal => + case _ if bucket.isDateHistogram => val functions = bucket.identifier.functions val interval: Option[DateHistogramInterval] = if (functions.size == 1) { diff --git a/bridge/src/test/scala/app/softnetwork/elastic/sql/QualifiedGroupKeyEmissionSpec.scala b/bridge/src/test/scala/app/softnetwork/elastic/sql/QualifiedGroupKeyEmissionSpec.scala new file mode 100644 index 00000000..c38cf69e --- /dev/null +++ b/bridge/src/test/scala/app/softnetwork/elastic/sql/QualifiedGroupKeyEmissionSpec.scala @@ -0,0 +1,339 @@ +/* + * 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 + +import app.softnetwork.elastic.sql.bridge._ +import app.softnetwork.elastic.sql.parser.Parser +import app.softnetwork.elastic.sql.query._ +import app.softnetwork.elastic.sql.schema.{Column, Table => SchemaTable} +import app.softnetwork.elastic.sql.`type`.SQLTypes +import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.time.ZonedDateTime +import scala.jdk.CollectionConverters._ + +/** Story IDENT-4 -- what Elasticsearch is SENT for a qualified GROUP BY key filter. + * + * RENDER level, at both resolutions: a single-table statement is resolved once at an index-pattern + * venue (and in a materialized view) and twice at a concrete-index one; a JOIN only by extensions + * and CTAS. The RUN-level truth is `GroupByCompletenessSpec`'s IDENT-4 rows. + */ +class QualifiedGroupKeyEmissionSpec extends AnyFlatSpec with Matchers { + + implicit def timestamp: Long = ZonedDateTime.parse("2025-12-31T00:00:00Z").toInstant.toEpochMilli + + private val mapper = new ObjectMapper() + + private val orders: SchemaTable = SchemaTable( + "orders", + columns = List( + Column("id", SQLTypes.Keyword), + Column("did", SQLTypes.Keyword), + Column("category", SQLTypes.Keyword), + Column("amount", SQLTypes.Int), + Column("d", SQLTypes.Date) + ) + ) + + private def accepted(sql: String): SingleSearch = + Parser(sql) match { + case Right(s: SingleSearch) => s + case other => fail(s"[$sql] expected an ACCEPTED SingleSearch, got $other") + } + + /** The COLLECTED form of [[accepted]]: a refusal is a wrong CELL, never an abort -- these tests + * read as a per-cell map. + */ + private def parsed(sql: String): Either[String, SingleSearch] = + Parser(sql) match { + case Right(s: SingleSearch) => Right(s) + case Right(other) => Left(s"not a SingleSearch: $other") + case Left(err) => Left(err.msg) + } + + private def query(s: SingleSearch): String = { + val request: ElasticSearchRequest = s + request.query + } + + private def walk(aggs: JsonNode, prefix: String): Seq[(String, JsonNode)] = + Option(aggs).toSeq.flatMap(_.fields().asScala.toSeq).flatMap { e => + val path = if (prefix.isEmpty) e.getKey else s"$prefix>${e.getKey}" + (path -> e.getValue) +: walk(e.getValue.get("aggs"), path) + } + + /** Every aggregation, keyed by its PATH (a self-join's two levels share a name). */ + private def aggregations(json: String): Seq[(String, JsonNode)] = + walk(mapper.readTree(json).get("aggs"), "") + + /** Every `terms` level's include, by path. */ + private def includes(json: String): Seq[(String, Option[String])] = + aggregations(json).collect { + case (path, body) if body.has("terms") => + path -> Option(body.get("terms").get("include")).map(_.toString) + } + + private val spellings = Seq( + "category" -> "category", + "category" -> "a.category", + "a.category" -> "category", + "a.category" -> "a.category" + ) + + /** How this bridge writes a ONE-value include (the es6 copy: `["x","x"]`, IDENT-3's wire rule). + */ + private val oneValueInclude = """["x"]""" + + private def keyed(groupBy: String, having: String): String = + s"SELECT COUNT(*) AS c, MAX(amount) AS mx FROM orders a GROUP BY $groupBy HAVING $having = 'x'" + + "IDENT-4: a key filter" should "reach Elasticsearch byte-identical to the bare spelling's, at both resolutions" in { + val control = accepted(keyed("category", "category")) + val r1Bare = query(control) + val r2Bare = query(control.update(Some(orders))) + // non-vacuity: the control carries the filter AND the metric on its level + includes(r1Bare) shouldBe Seq("category" -> Some(oneValueInclude)) + aggregations(r1Bare).map(_._1) should contain("category>mx") + val wrong = spellings.flatMap { case (k, h) => + parsed(keyed(k, h)) match { + case Left(reason) => Seq(s"[$k / $h] refused: $reason") + case Right(s) => + Seq( + Some(query(s)).filter(_ != r1Bare).map(q => s"[$k / $h] R1:\n $q\n bare $r1Bare"), + Some(query(s.update(Some(orders)))) + .filter(_ != r2Bare) + .map(q => s"[$k / $h] R2:\n $q\n bare $r2Bare") + ).flatten + } + } + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + "a self-join's key filter" should "sit on the ONE level its correlation names, at both resolutions" in { + val sql = (alias: String) => + "SELECT COUNT(*) AS c, MAX(o.amount) AS mx FROM orders o JOIN orders p ON o.did = p.id " + + s"GROUP BY o.category, p.category HAVING $alias.category = 'x'" + val wrong = for { + (alias, filtered) <- Seq("o" -> "category", "p" -> "category>category") + (label, statement) <- Seq( + "R1" -> parsed(sql(alias)), + "R2" -> parsed(sql(alias)).map(_.updateAll(Map("orders" -> orders))) + ) + problem <- statement match { + case Left(reason) => Seq(s"[$alias.category, $label] refused: $reason") + case Right(s) => + val actual = includes(query(s)) + val expected = Seq("category", "category>category").map(p => + p -> (if (p == filtered) Some(oneValueInclude) else None) + ) + if (actual != expected) Seq(s"[$alias.category, $label] $actual, expected $expected") + else Nil + } + } yield problem + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + "GROUP BY over a qualified column" should + "emit byte-identically to its unqualified twin, at both resolutions" in { + // The twin at BOTH resolutions: at one resolution the qualified body used to lack the metric + // sub-aggregation (a qualified key's metrics staying on its level, through the alias-substituted copy). The metric makes it observable. + val wrong = Seq( + "SELECT o.category AS cat, COUNT(*) AS c FROM orders o GROUP BY cat HAVING cat = 'x'" -> + "SELECT category AS cat, COUNT(*) AS c FROM orders GROUP BY cat HAVING cat = 'x'", + "SELECT o.category AS cat, COUNT(*) AS c, MAX(amount) AS mx FROM orders o GROUP BY cat HAVING cat = 'x'" -> + "SELECT category AS cat, COUNT(*) AS c, MAX(amount) AS mx FROM orders GROUP BY cat HAVING cat = 'x'" + ).flatMap { case (q, b) => + val (qualified, twin) = (accepted(q), accepted(b)) + Seq( + Some(query(qualified)) + .filter(_ != query(twin)) + .map(x => s"[$q] R1:\n $x\n twin ${query(twin)}"), + Some(query(qualified.update(Some(orders)))) + .filter(_ != query(twin.update(Some(orders)))) + .map(x => s"[$q] R2:\n $x\n twin ${query(twin.update(Some(orders)))}") + ).flatten + } + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + "a qualified GROUP BY key with a metric" should + "keep the metric on its level after ONE resolution -- the body an index pattern sends" in { + // The metric's path used to be the parsed spelling `a.category`, its level's `category`, and a + // metric no level claims is dropped under a GROUP BY. + val wrong = Seq( + "SELECT a.category, MAX(amount) AS mx FROM orders a GROUP BY a.category", + "SELECT category, MAX(amount) AS mx FROM orders a GROUP BY a.category", + "SELECT a.category AS k, MAX(amount) AS mx FROM orders a GROUP BY 1" + ).flatMap { sql => + val paths = + aggregations(query(accepted(sql))).map( + _._1 + ) // accepted before too: a qualified key's metrics staying on its level is a RENDER defect + if (paths.exists(_.endsWith(">mx"))) None + else Some(s"[$sql] no metric under its level: $paths") + } + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + // --------------------------------------------------------------------------------------------- + // The chain of a qualified column resolving with it, the date-histogram key-filter refusal's parse arm and the WHERE temporal comparison, at R1 on this bridge + // --------------------------------------------------------------------------------------------- + + /** Every `field` and script text under the aggregations -- what a key reads. */ + private def keyReads(json: String): Seq[String] = { + def texts(node: JsonNode): Seq[String] = + if (node == null) Nil + else if (node.isObject) + node.fields().asScala.toSeq.flatMap { e => + if (Set("field", "source", "inline", "script")(e.getKey) && e.getValue.isTextual) + Seq(e.getValue.asText) + else texts(e.getValue) + } + else if (node.isArray) node.elements().asScala.toSeq.flatMap(texts) + else Nil + texts(mapper.readTree(json).get("aggs")) + } + + private val dateSchema: SchemaTable = + SchemaTable( + "t", + columns = + List(Column("d", SQLTypes.Date), Column("e", SQLTypes.Date), Column("s", SQLTypes.Keyword)) + ) + + "a function of a qualified column as the GROUP BY key" should + "read the RESOLVED field after ONE resolution -- every registry function key, projected or not" in { + // A key script naming the PARSED column (`doc['q.d']`) reads a field no mapping has: wrong or + // failed for every mapping. Judged over every function key of the registry (story IDENT-1's + // population), over `FROM t q`, the key projected and not (the GROUPING, not the metrics). + val rows = for { + (spelling, key, _) <- QualifiedKeyPopulation.functionKeys + k = key.replace("t.d", "q.d") + sql <- Seq( + s"SELECT COUNT(*) AS c FROM t q GROUP BY $k", + s"SELECT $k AS k, COUNT(*) AS c FROM t q GROUP BY $k" + ) + s <- parsed(sql).toOption.toSeq + } yield (spelling, sql, keyReads(query(s))) + val stale = rows.filter(r => r._3.exists(read => read.contains("'q.d'") || read == "q.d")) + val resolved = rows.filter(_._3.exists(_.contains("doc['d']"))) + info( + s"${rows.size} R1 renders · ${resolved.size} key scripts read doc['d'] · ${stale.size} read the parsed q.d" + ) + resolved should not be empty // non-vacuity: the stale-argument members render a key script + withClue(stale.map(r => s"[${r._1}] ${r._2}: ${r._3.mkString(" ;; ")}").mkString("\n")) { + stale shouldBe empty + } + } + + "a function key whose output is temporal without a mapping" should + "be emitted at R1 as a date histogram -- the premise of the date-histogram key-filter refusal parse arm" in { + // The population reaching the parse arm: every registry function key (bare `d`, depth 1) whose + // level is temporal with no mapping. Each must reach Elasticsearch as a date histogram, the + // level that has no key filter -- else the arm would refuse a filter that could be applied. + val rows = QualifiedKeyPopulation.functionKeys.flatMap { case (spelling, _, bare) => + parsed(s"SELECT COUNT(*) AS c FROM t GROUP BY $bare").toOption.flatMap { s => + s.buckets.headOption.filter(_.isDateHistogram).map { _ => + val level = Option(mapper.readTree(query(s)).get("aggs")) + .flatMap(a => Option(a.elements()).filter(_.hasNext).map(_.next())) + ( + spelling, + bare, + level.exists(_.has("date_histogram")), + parsed( + s"SELECT COUNT(*) AS c FROM t GROUP BY $bare HAVING $bare = '2024-01-01'" + ) + ) + } + } + } + info(s"${rows.size} temporal-output keys reach the parse arm") + rows should not be empty + val wrong = rows.flatMap { case (spelling, bare, histogram, filtered) => + Seq( + if (histogram) None + else Some(s"[$spelling] GROUP BY $bare is not emitted as a date histogram"), + filtered match { + case Left(reason) if reason.startsWith("HAVING cannot filter a DATE grouping level") => + None + case other => Some(s"[$spelling] GROUP BY $bare HAVING $bare = '2024-01-01': $other") + } + ).flatten + } + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + "a temporal WHERE comparison of a GROUP BY key column" should + "compare the two dates as dates, qualified or bare, whether or not the column is grouped" in { + // A WHERE comparison reads documents, never a bucket key: whether its column is also a GROUP BY + // key must not decide its Painless. It used to take the numeric fallback -- a raw `>` on two + // dates, which Painless refuses -- exactly when the column was attached to its key. So each + // statement emits the WHERE its ungrouped twin emits, and a comparison with a date the statement + // computes (`DATE_ADD`) -- or, once the mapping is known, with a date column -- is a date one. + val statements = Seq( + "SELECT COUNT(*) AS c FROM t q WHERE q.d > DATE_ADD(q.e, INTERVAL 1 DAY) GROUP BY q.d", + "SELECT COUNT(*) AS c FROM t WHERE d > DATE_ADD(e, INTERVAL 1 DAY) GROUP BY d", + "SELECT COUNT(*) AS c FROM t q WHERE q.d > q.e GROUP BY q.d", + "SELECT COUNT(*) AS c FROM t WHERE d > e GROUP BY d" + ) + val wrong = for { + sql <- statements + // R1 only: at R2 this bridge renders a column-vs-column date comparison as a `range` on the + // other column's NAME, a pre-existing translation defect outside this story + (label, statement) <- Seq("R1" -> parsed(sql)) + ungrouped = sql.substring(0, sql.indexOf(" GROUP BY")) + problem <- (statement, parsed(ungrouped)) match { + case (Right(s), Right(control)) => + val where = Option(mapper.readTree(query(s)).get("query")).map(_.toString).getOrElse("") + val twin = + Option(mapper.readTree(query(control)).get("query")).map(_.toString).getOrElse("") + val dated = sql.contains("DATE_ADD") || label == "R2" + Seq( + if (!dated || where.contains("isAfter")) None + else Some(s"[$label] $sql: no date comparison in $where"), + // the ungrouped twin, at the same resolution: the same comparison, grouped or not + if (label != "R1" || where == twin) None + else Some(s"[$label] $sql: $where, ungrouped $twin") + ).flatten + case other => Seq(s"[$label] $sql: $other") + } + } yield problem + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + // The date-histogram key-filter refusal -- the ONE derivation both bridges and the classifier read, tied in BOTH directions: the + // levels the date-histogram key-filter refusal refuses and the ones it ACCEPTS (a key filter that reaches a `terms` level). + "Bucket.isDateHistogram" should "be true exactly for the levels the bridge emits as a date_histogram" in { + val observations = for { + sql <- Seq( + "SELECT COUNT(*) AS c FROM orders GROUP BY DATE_TRUNC(d, MONTH)", + "SELECT COUNT(*) AS c FROM orders GROUP BY YEAR(d)", + "SELECT COUNT(*) AS c FROM orders GROUP BY category", + "SELECT COUNT(*) AS c FROM orders GROUP BY d", + "SELECT COUNT(*) AS c FROM orders GROUP BY YEAR(d) HAVING YEAR(d) = 2024", + "SELECT COUNT(*) AS c FROM orders a GROUP BY a.d HAVING a.d = '2024-01-03'" + ) + s <- Seq(accepted(sql), accepted(sql).update(Some(orders))) + level = aggregations(query(s)).head._2 + } yield (sql, s.buckets.head.isDateHistogram, level.has("date_histogram")) + observations.filter(o => o._2 != o._3) shouldBe empty + observations.map(_._2).distinct.sorted shouldBe Seq(false, true) // both kinds observed + } + + "IDENT-4 on Elasticsearch 6.8.23 (in process)" should + "keep the same keys for every spelling of a key filter, at both resolutions" in { + import Es68TermsFilterOracle._ + val keys = Seq("x", "y", "xa", "z") + val ops: Seq[(String, String => String, Set[String])] = Seq( + ("=", k => s"$k = 'x'", Set("x")), + ("<>", k => s"$k <> 'x'", Set("y", "xa", "z")), + ("IN", k => s"$k IN ('x', 'y')", Set("x", "y")), + ("NOT IN", k => s"$k NOT IN ('x', 'y')", Set("xa", "z")), + ("LIKE", k => s"$k LIKE 'x%'", Set("x", "xa")), + ("NOT LIKE", k => s"$k NOT LIKE 'x%'", Set("y", "z")), + ("RLIKE", k => s"$k RLIKE 'x.*'", Set("x", "xa")), + ("MATCH", k => s"MATCH ($k) AGAINST ('x')", Set("x")) + ) + val wrong = for { + (op, predicate, kept) <- ops + (groupBy, having) <- spellings + sql = + s"SELECT COUNT(*) AS c, MAX(amount) AS mx FROM orders a GROUP BY $groupBy HAVING ${predicate(having)}" + (label, statement) <- Seq( + "R1" -> parsed(sql), + "R2" -> parsed(sql).map(_.update(Some(orders))) + ) + problem <- statement match { + case Left(reason) => Seq(s"[$op | $groupBy / $having | $label] refused: $reason") + case Right(s) => + val verdict = keptBy(termsBody(query(s), "category"), KeywordKey, keys) + if (verdict != Kept(kept)) + Seq(s"[$op | $groupBy / $having | $label] $verdict, expected $kept") + else Nil + } + } yield problem + withClue(s"${wrong.size} wrong:\n${wrong.mkString("\n")}\n") { wrong shouldBe empty } + } + + it should "keep every key on the level a self-join's filter does not name" in { + import Es68TermsFilterOracle._ + val sql = (alias: String) => + "SELECT COUNT(*) AS c, MAX(o.amount) AS mx FROM orders o JOIN orders p ON o.did = p.id " + + s"GROUP BY o.category, p.category HAVING $alias.category = 'x'" + val wrong = for { + (alias, filtered) <- Seq("o" -> "category", "p" -> "category>category") + (label, statement) <- Seq( + "R1" -> parsed(sql(alias)), + "R2" -> parsed(sql(alias)).map(_.updateAll(Map("orders" -> orders))) + ) + problem <- statement match { + case Left(reason) => Seq(s"[$alias.category, $label] refused: $reason") + case Right(s) => + termsBodies(query(s)).flatMap { case (path, body) => + val verdict = keptBy(body, KeywordKey, Seq("x", "y")) + val expected = if (path == filtered) Kept(Set("x")) else Kept(Set("x", "y")) + if (verdict != expected) + Seq(s"[$alias.category, $label, $path] $verdict, expected $expected") + else Nil + } + } + } yield problem + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + it should "keep IDENT-3's ES 6 refusal set on the qualified spelling (a pattern beside two values)" in { + // IDENT-3 contract item 4: a pattern KEPT beside two or more values REMOVED is inexpressible on + // Elasticsearch 6.8 -- its request parser rejects the mix (`IncludeExclude.merge`, HTTP 400). + // The es6 serializers refuse it by name before sending -- in all four spellings, at both + // resolutions. + val wrong = for { + (groupBy, having) <- spellings + sql = s"SELECT COUNT(*) AS c, MAX(amount) AS mx FROM orders a GROUP BY $groupBy " + + s"HAVING $having LIKE 'x%' AND $having NOT IN ('xa', 'xb')" + (label, statement) <- Seq( + "R1" -> parsed(sql), + "R2" -> parsed(sql).map(_.update(Some(orders))) + ) + problem <- statement match { + case Left(reason) => Seq(s"[$groupBy / $having | $label] refused at parse: $reason") + case Right(s) => + // the es6 serializer refuses the mix by name before sending (IDENT-3 contract item 4) + scala.util.Try(query(s)) match { + case scala.util.Failure(e: UnsupportedOperationException) + if String.valueOf(e.getMessage).contains("Elasticsearch before 7.11") => + Nil + case scala.util.Failure(e) => Seq(s"[$groupBy / $having | $label] threw $e") + case scala.util.Success(body) => + Seq(s"[$groupBy / $having | $label] sent $body, expected the serializer's refusal") + } + } + } yield problem + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } +} diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala index 8fcc5801..121c5cca 100644 --- a/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/package.scala @@ -2692,7 +2692,11 @@ package object sql { nested = true, limit = tuple._2._2, fieldAlias = request.fieldAliases.get(identifierName).orElse(fieldAlias), - bucket = request.bucketNames.get(identifierName).orElse(bucket), + // Story IDENT-4 -- ONE decision for all five branches: `SingleSearch.bucketFor`. + // In the three nested branches it IS the spelling lookup (a nested or UNNEST- + // qualified name has no identity), so only the qualified-plain and unqualified + // branches can answer differently. + bucket = request.bucketFor(this).orElse(bucket), nestedElement = nestedElement, bucketPath = bucketPath, col = columnSchema.flatMap(schema => schema.find(colName)), @@ -2707,7 +2711,7 @@ package object sql { name = s"${tuple._2._1}.$colName", limit = tuple._2._2, fieldAlias = request.fieldAliases.get(identifierName).orElse(fieldAlias), - bucket = request.bucketNames.get(identifierName).orElse(bucket), + bucket = request.bucketFor(this).orElse(bucket), bucketPath = bucketPath, col = columnSchema.flatMap(schema => schema.find(colName)), table = table @@ -2718,7 +2722,7 @@ package object sql { .copy( tableAlias = Some(tableAlias), fieldAlias = request.fieldAliases.get(identifierName).orElse(fieldAlias), - bucket = request.bucketNames.get(identifierName).orElse(bucket), + bucket = request.bucketFor(this).orElse(bucket), bucketPath = bucketPath, col = columnSchema.flatMap(schema => schema.find(name)), table = table @@ -2726,15 +2730,21 @@ package object sql { .withFunctions(this.updateFunctions(request)) case _ => val colName = parts.tail.mkString(".") - this.copy( - tableAlias = Some(tableAlias), - name = colName, - fieldAlias = request.fieldAliases.get(identifierName).orElse(fieldAlias), - bucket = request.bucketNames.get(identifierName).orElse(bucket), - bucketPath = bucketPath, - col = columnSchema.flatMap(schema => schema.find(colName)), - table = table - ) + this + .copy( + tableAlias = Some(tableAlias), + name = colName, + fieldAlias = request.fieldAliases.get(identifierName).orElse(fieldAlias), + bucket = request.bucketFor(this).orElse(bucket), + bucketPath = bucketPath, + col = columnSchema.flatMap(schema => schema.find(colName)), + table = table + ) + // Story IDENT-4 -- the chain resolves with the column, as in every other branch. + // Without it every function in a qualified column's chain kept the PARSED receiver + // (`doc['t.d']`) until a second `update()` -- which an index-pattern search, a + // materialized view and a watcher never perform. + .withFunctions(this.updateFunctions(request)) } } else { // An UN-QUALIFIED name. It resolves against the MAIN table, whatever that is — and since @@ -2745,7 +2755,7 @@ package object sql { this .copy( fieldAlias = request.fieldAliases.get(identifierName).orElse(fieldAlias), - bucket = request.bucketNames.get(identifierName).orElse(bucket), + bucket = request.bucketFor(this).orElse(bucket), bucketPath = bucketPath, col = columnSchema.flatMap(schema => schema.find(name)) ) diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/query/From.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/query/From.scala index 5b25464d..55ed1e39 100644 --- a/sql/src/main/scala/app/softnetwork/elastic/sql/query/From.scala +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/query/From.scala @@ -706,6 +706,17 @@ case class From(tables: Seq[Table]) extends Updateable { */ lazy val mainTableKey: String = aliasKey(mainTable.name, mainTable.qualifiedName) + /** Story IDENT-4 -- the CORRELATION NAME of the main table: its SQL alias, else its bare name + * (the value [[tableAliases]] records for it). What an UNQUALIFIED column is correlated with, + * since an unqualified name resolves against the main table (`GenericIdentifier.update`). + * + * 🔴 NOT [[mainTableKey]]. That is the alias-map KEY -- the INDEX in `aliasKey`'s language -- + * and both legs of a self-join (`FROM orders a JOIN orders b`) share it. Something comparing two + * COLUMNS of one statement keys on the correlation; only something addressing a MAPPING keys on + * the index. + */ + lazy val mainTableCorrelation: String = effectiveAlias(mainTable) + /** 🔴 FIX (story BIDC-8, softclient4es-arrow#144) — alias -> table key, built DIRECTLY and * LOSSLESS. It used to be `tableAliases.map(_.swap)`. * diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/query/GroupBy.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/query/GroupBy.scala index dda0a3c2..79098281 100644 --- a/sql/src/main/scala/app/softnetwork/elastic/sql/query/GroupBy.scala +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/query/GroupBy.scala @@ -16,7 +16,7 @@ package app.softnetwork.elastic.sql.query -import app.softnetwork.elastic.sql.`type`.SQLType +import app.softnetwork.elastic.sql.`type`.{SQLTemporal, SQLType} import app.softnetwork.elastic.sql.function.aggregate._ import app.softnetwork.elastic.sql.operator._ import scala.util.{DynamicVariable, Try} @@ -286,6 +286,32 @@ case class Bucket( lazy val path: String = sourceBucket + /** Story IDENT-4 -- do two buckets group on the SAME key? The ONE comparison every "which level + * does this HAVING leaf address" question reads: the three include sites of `Where.scala` + * (`Expression`, `InExpr`, `MatchCriteria`) and `SingleSearch.keyBucketOf`. + * + * It used to be `name == other.name`, and [[name]] is the DISPLAY name -- the SELECT alias, else + * the column. Three ways that is not an identity: + * - a SELF-JOIN's two levels (`GROUP BY o.category, p.category`) are both named `category` + * after one resolution, so a condition on one correlation filtered BOTH levels; + * - two FUNCTIONS of one column (`GROUP BY YEAR(d), MONTH(d)`) are both named `d` when the + * SELECT does not project them, so `HAVING MONTH(d) = 1` also filtered the YEAR level; + * - a copy recorded before resolution (`a.category`) never equalled its resolved level + * (`category`), which is why the qualified spelling was refused. + * + * So the key is the display name AND the correlation AND the grouped expression -- strings only, + * never `Identifier.equals`, which compares a doc access, not a grouping key. + * + * A NESTED level keeps the display-name comparison: its attached copy is recorded from the + * parsed spelling (no alias, no nested element) while the level is resolved, and only the + * display name agrees between them (`sourceBucket`). + */ + def sameKeyAs(other: Bucket): Boolean = + name == other.name && (nested || other.nested || ( + identifier.tableAlias == other.identifier.tableAlias && + identifier.identifierName == other.identifier.identifierName + )) + lazy val nestedPath: String = { identifier.nestedElement match { case Some(ne) => ne.nestedPath @@ -295,6 +321,12 @@ case class Bucket( override def out: SQLType = identifier.out + /** Story IDENT-4 -- does this level reach Elasticsearch as a `date_histogram`? The ONE derivation + * both bridges' emission AND the HAVING classifier read: a `date_histogram` carries no include / + * exclude, so a key filter addressed to this level could not be applied. + */ + def isDateHistogram: Boolean = out.isInstanceOf[SQLTemporal] + /** A bucket with NO FIELD NAME must be scripted, or nothing describes it. * * An Elasticsearch `terms` aggregation must specify a `field` or a `script`. A bucket resolved diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/query/Where.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/query/Where.scala index b28e36ca..36e1f073 100644 --- a/sql/src/main/scala/app/softnetwork/elastic/sql/query/Where.scala +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/query/Where.scala @@ -701,7 +701,9 @@ sealed trait Expression extends FunctionChain with ElasticFilter with Criteria { not: Boolean, bucketIncludesExcludes: BucketIncludesExcludes ): BucketIncludesExcludes = { - identifier.bucket.find(_.name == bucket.name) match { + // Story IDENT-4 -- the SAME grouping key, not merely the same display name (`Bucket.sameKeyAs`): + // a display name is shared by a self-join's two levels and by two functions of one column. + identifier.bucket.find(_.sameKeyAs(bucket)) match { case Some(_) => operator match { // These feed the ES `terms` include/exclude and its regex form, so they need the value @@ -1404,8 +1406,12 @@ sealed trait Expression extends FunctionChain with ElasticFilter with Criteria { // milliseconds -- and [[bucketPipelineCheck]] converts a temporal value to the same unit, so the // two are compared as numbers. The `java.time` methods below would be called on a `Double` // (`dynamic method [java.lang.Double, isAfter/1] not found`). + // + // Story IDENT-4 -- NOT "the column is a GROUP BY key" (`hasBucket`): a WHERE comparison reads + // documents, never a bucket key, so it took the numeric fallback (a raw `>` on two dates, + // which Painless refuses) whenever its column happened to be grouped, and only then. val readsMetric = - isAggregation || hasBucket || (context.isEmpty && (referencesBucketMetric || reducesDateMetrics)) + isAggregation || (context.isEmpty && (referencesBucketMetric || reducesDateMetrics)) op match { case _: ComparisonOperator => // 🔴 Story BIDC-8 AD-9 — dispatch on the EFFECTIVE operator, i.e. with a `NOT` folded in, @@ -1857,7 +1863,7 @@ sealed trait Expression extends FunctionChain with ElasticFilter with Criteria { */ val utcComparison: Boolean = context.exists(c => !c.isProcessor && !c.isTransform && c.target.jodaCompatibleDates) && - valueType.isInstanceOf[SQLTemporal] && !isAggregation && !hasBucket && + valueType.isInstanceOf[SQLTemporal] && !isAggregation && comparisonTargetType == SQLTypes.Timestamp def instantOperand(rendered: SQLType, readsDocument: Boolean): Boolean = utcComparison && readsDocument && @@ -2323,7 +2329,7 @@ case class InExpr[R, +T <: Value[R]]( not: Boolean, bucketIncludesExcludes: BucketIncludesExcludes ): BucketIncludesExcludes = { - identifier.bucket.find(_.name == bucket.name) match { + identifier.bucket.find(_.sameKeyAs(bucket)) match { case Some(_) => if ((!not && maybeNot.isEmpty) || (not && maybeNot.isDefined)) // The VALUE, never its SQL rendering -- the rule `Expression.includes` states on its EQ @@ -3097,7 +3103,7 @@ case class MatchCriteria( not: Boolean, bucketIncludesExcludes: BucketIncludesExcludes ): BucketIncludesExcludes = { - identifier.bucket.find(_.name == bucket.name) match { + identifier.bucket.find(_.sameKeyAs(bucket)) match { case Some(_) => if ((!not && maybeNot.isEmpty) || (not && maybeNot.isDefined)) // The value, not its rendering — see the include/exclude note above. 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 f8832d54..9dd4d6a7 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 @@ -437,6 +437,32 @@ package object query { def minScore(score: Double): SelectStatement = this.copy(score = Some(score)) } + /** Story IDENT-4 -- one bucket as [[SingleSearch.bucketFor]] indexes it. + * + * A plain class on purpose: it is never compared, only searched, and a case class's structural + * `equals` would reach `Identifier.equals` through the bucket -- the collapse this story must + * not depend on. + * + * @param bucket + * the lookup copy -- qualifier-RESOLVED when it is a GROUP BY key + * @param keys + * every spelling it is looked up by, in the order `bucketNames` has always used + * @param aliasKeys + * the subset of `keys` that are ALIASES (they name the bucket by definition) + * @param groupByKey + * a GROUP BY key (`eq` membership), not a window PARTITION BY bucket + * @param identity + * (correlation as written, grouped expression) -- `None` for a nameless or a nested key, which + * keeps the spelling lookup + */ + private[query] final class KeyedBucket( + val bucket: Bucket, + val keys: Seq[String], + val aliasKeys: Set[String], + val groupByKey: Boolean, + val identity: Option[(Option[String], String)] + ) + case class SingleSearch( select: Select = Select(), from: From, @@ -616,9 +642,10 @@ package object query { * 🔴 OBLIGATION, not a reassurance: anything that looks a bucket up in this map must speak one * of those two key languages. Adding a resolution rule here (the ordinal and alias arms below) * without indexing BOTH spellings silently disables one of the two lookups -- which is exactly - * how `GROUP BY pays` came to emit `terms { field: "pays" }` on a non-existent field. The - * consumers are `sql/.../package.scala` (`Identifier.update`, 6 sites) and - * `sql/.../function/aggregate/package.scala` (`WindowFunction.update`, 2 sites). + * how `GROUP BY pays` came to emit `terms { field: "pays" }` on a non-existent field. Its + * consumers are the `BucketPath` of `Identifier.update` (VALUES) and `WindowFunction.update` + * (KEYS). The five attachment sites of `Identifier.update` ask [[bucketFor]] instead, which + * reads [[keyedBuckets]], the list this map is derived from. * * Issue #253's companion defect: ordinal-ness used to be sniffed with `"\\d+".r.findFirstIn` * on the RENDERED name, guarded only by "the name contains no space", so ANY name carrying a @@ -634,47 +661,194 @@ package object query { * builds LOOKUP KEYS; `Bucket.update` owns the resolution that reaches the emitted query, and * both route through `Bucket.selectItem` / `Bucket.aliasItem` so they cannot disagree. */ - lazy val bucketNames: ListMap[String, Bucket] = ListMap(buckets.flatMap { b => + lazy val bucketNames: ListMap[String, Bucket] = + ListMap(keyedBuckets.flatMap(k => k.keys.map(_ -> k.bucket)): _*) + + /** Story IDENT-4 -- every bucket as the lookups index it, in `buckets` order, NEVER collapsed: + * one entry per bucket instance. Nothing here is a `Map`, a `Set` or a `distinct` over + * buckets, so two buckets that print alike -- a self-join's `o.category` and `p.category`, + * both named `category` after one resolution -- stay two entries. (A `ListMap` keyed by name + * cannot be the search set: after one resolution it holds one of the two.) + * + * `keys` / the copy are exactly what [[bucketNames]] held before, with ONE change: a GROUP BY + * key's copy is qualifier-RESOLVED ([[tableQualifierOf]], the rule `GenericIdentifier.update` + * applies), so a copy made from the parsed spelling (`a.category`) carries the alias and the + * column its resolved level carries (`a` / `category`). That is what lets an attachment + * recorded at the FIRST resolution be compared with the level it addresses, and what puts a + * qualified key's metrics on their level after ONE resolution (the `BucketPath` reads these + * copies). + * + * ⚠️ Still NOT `.update(this)`-d, for the reason [[bucketNames]] gives: `Identifier.update` + * forces this very value. + */ + private[query] lazy val keyedBuckets: Seq[KeyedBucket] = buckets.map { b => val name = b.identifier.identifierName + val groupByKey = groupBy.exists(_.buckets.exists(_ eq b)) val resolved: Option[Identifier] = (b.ordinal match { case Some(position) => Bucket.selectItem(position, this) case None => Bucket.aliasItem(b.identifier, this) }).map(_.identifier) - resolved match { + val (updated, keys, aliasKeys) = resolved match { case Some(identifier) => val resolvedName = identifier.identifierName // Carry the SELECT alias so this copy's `name` equals the real bucket's -- see the - // OBLIGATION above; `Identifier.update` reads the same `fieldAliases` map. + // OBLIGATION on `bucketNames`; `Identifier.update` reads the same `fieldAliases` map. val display = fieldAliases.get(resolvedName) val named = identifier match { case g: GenericIdentifier => g.copy(fieldAlias = display.orElse(g.fieldAlias)) case other => other } - val updated = b.copy(identifier = named) - ListMap( - (Seq( - name -> updated, // the spelling the statement was written with - resolvedName -> updated // the resolved column name - ) ++ display.map(_ -> updated)): _* // the OUTPUT name a HAVING or WHERE would use + ( + b.copy(identifier = named), + // the spelling the statement was written with, the resolved column name, and the + // OUTPUT name a HAVING or WHERE would use + (Seq(name, resolvedName) ++ display).distinct, + Set(name) ++ display ) case None => // The bucket was NOT substituted -- but it still has to be named the way the real bucket // is, or `Expression.includes` (which compares by DISPLAY name) stops matching and a // HAVING is silently dropped. `SELECT category AS cat ... GROUP BY category HAVING cat // <> 'x'` lost its `exclude` for exactly this reason: the copy was named `category` while - // the bucket that reaches Elasticsearch is named `cat`. Broken on main too, but leaving - // it made whether a HAVING is honoured depend on whether the GROUP BY spelled the alias - // or the column, which is worse than uniformly broken. + // the bucket that reaches Elasticsearch is named `cat`. val display = fieldAliases.get(name) val named = b.identifier match { case g: GenericIdentifier => g.copy(fieldAlias = display.orElse(g.fieldAlias)) case other => other } - val updated = b.copy(identifier = named) - ListMap((Seq(name -> updated) ++ display.map(_ -> updated)): _*) + (b.copy(identifier = named), (Seq(name) ++ display).distinct, display.toSet) + } + val bucket = if (groupByKey) withTableQualifierResolved(updated) else updated + new KeyedBucket(bucket, keys, aliasKeys, groupByKey, keyIdentityOf(bucket.identifier)) + } + + /** Story IDENT-4 -- THE decision "which bucket does this identifier name?", asked by every + * `GenericIdentifier.update` branch and recorded in its `bucket` attachment: emission holds + * only a `Criteria` and a `Bucket` and can decide nothing later. + * + * It used to be `bucketNames.get(identifierName)`, a lookup by SPELLING, which (1) missed a + * qualified key spelled differently from its bucket (`HAVING a.category` against `GROUP BY + * category`, or the reverse), so the qualified spelling was refused; and (2) after one + * resolution, when a self-join's two buckets are both spelled `category`, answered the LAST + * one for either correlation. + * + * Three rules, in order; each keeps `today` (the spelling lookup) unless it is provably the + * wrong column: + * - R-a: `today` stands when the identifier named it by an ALIAS -- which keeps `SELECT + * o.category AS cat ... GROUP BY cat HAVING cat = 'x'` exactly as it is; + * - R-b: otherwise the GROUP BY key with the SAME identity, else the one it is + * main-equivalent to (an unqualified name is the main table's column, as everywhere in + * `update`). GROUP BY keys only: a window PARTITION BY bucket never takes this route; + * - R-c: otherwise `today`, when its identity is absent or compatible; else nothing -- the + * predicate names no grouping key, and the HAVING rules refuse it by name. + * + * An identifier with no identity (nameless, nested, an UNNEST column) keeps `today`. When + * neither the identifier nor any key is qualified, `today` is returned directly: its answer + * cannot move, and it costs nothing more. + * + * Strings and `eq` only -- nothing here reads `Identifier.equals`. + */ + private[sql] def bucketFor(identifier: Identifier): Option[Bucket] = + if (keyedBuckets.isEmpty) None + else { + val spelling = identifier.identifierName + // the last entry wins -- exactly what `bucketNames.get(spelling)` answered + val today = keyedBuckets.reverseIterator.find(_.keys.contains(spelling)) + val chosen = + if (!carriesQualifier && !isQualified(identifier)) today + else + keyIdentityOf(identifier) match { + case None => today + case Some(leaf) => + today + .filter(_.aliasKeys.contains(spelling)) + .orElse(identityRoute(leaf)) + .orElse(today.filter(_.identity.forall(compatible(_, leaf)))) + } + chosen.map(_.bucket) } - }: _*) + + /** Story IDENT-4 -- the identity of a column reference for the bucket lookup: its CORRELATION + * as written (the table alias; `None` when unqualified) and the grouped EXPRESSION spelled + * over the bare column. `None` when there is nothing to compare: a nameless carrier (a + * constant, a string / math / CASE call) or a nested (UNNEST) column. It reads the parsed + * spelling (`a.d`, no alias yet) and the resolved one (`d`, alias `a`) alike -- which is what + * makes the two resolutions agree. + * + * 🔴 The correlation, never the INDEX (`Identifier.table`): both legs of a self-join resolve + * to one index, and only the alias tells them apart. + */ + private[query] def keyIdentityOf(identifier: Identifier): Option[(Option[String], String)] = + if (identifier.name.isEmpty || identifier.nested) None + else + identifier.tableAlias match { + case Some(alias) => Some(Some(alias) -> identifier.identifierName) + case None => + if (unnestQualified(identifier.name)) None + else + tableQualifierOf(identifier.name) match { + case Some((alias, column, _)) => Some(Some(alias) -> identifier.toSQL(column)) + case None => Some(None -> identifier.identifierName) + } + } + + /** Story IDENT-4 -- `(alias, column, table key)` when a dotted name starts with a correlation + * this FROM declares, exactly as `GenericIdentifier.update`'s qualified-plain branch splits it + * (first dot; the qualifier looked up in [[aliasesToTable]]; the rest is the column). An + * UNNEST alias is excluded: `update` sends that name down its nested branches. A table's own + * name is no qualifier once the table is aliased (it is not in `aliasesToTable`). + * + * ⚠️ A MIRROR of `update`'s rule, which cannot be called from here (it forces + * [[bucketNames]]): `QualifiedGroupKeySpec` compares the two over a derived name population. + */ + private[query] def tableQualifierOf(name: String): Option[(String, String, String)] = { + val dot = name.indexOf('.') + if (dot <= 0 || dot == name.length - 1) None + else { + val qualifier = name.substring(0, dot) + if (unnestAliases.contains(qualifier)) None + else aliasesToTable.get(qualifier).map(key => (qualifier, name.substring(dot + 1), key)) + } + } + + private[this] def withTableQualifierResolved(b: Bucket): Bucket = + b.identifier match { + case g: GenericIdentifier if !g.nested && g.tableAlias.isEmpty => + tableQualifierOf(g.name) match { + case Some((alias, column, key)) => + b.copy(identifier = + g.copy(name = column, tableAlias = Some(alias), table = Some(key)) + ) + case None => b + } + case _ => b + } + + private[this] def identityRoute(leaf: (Option[String], String)): Option[KeyedBucket] = { + val keys = keyedBuckets.filter(_.groupByKey) + keys + .find(_.identity.contains(leaf)) + .orElse(keys.find(_.identity.exists(compatible(_, leaf)))) + } + + private[this] def compatible( + a: (Option[String], String), + b: (Option[String], String) + ): Boolean = + a._2 == b._2 && + a._1.getOrElse(from.mainTableCorrelation) == b._1.getOrElse(from.mainTableCorrelation) + + private[this] def unnestQualified(name: String): Boolean = { + val dot = name.indexOf('.') + dot > 0 && unnestAliases.contains(name.substring(0, dot)) + } + + private[this] def isQualified(identifier: Identifier): Boolean = + identifier.tableAlias.isDefined || tableQualifierOf(identifier.name).isDefined + + private[this] lazy val carriesQualifier: Boolean = + keyedBuckets.exists(_.identity.exists(_._1.isDefined)) var unnests: scala.collection.mutable.Map[String, Unnest] = { val map = from.unnests.map(u => u.alias.map(_.alias).getOrElse(u.name) -> u).toMap @@ -1344,19 +1518,62 @@ package object query { } private[query] def keyBucketOf(leaf: Identifier): Option[Bucket] = { - // 🔴 Resolve through THIS statement's bucket list, never through `Identifier.bucket`, which - // is a COPY attached during `update()` and deliberately NOT re-updated (the story-21.3 / #253 - // desync class) -- MEASURED reporting `nestedElement = None` for a bucket the statement lists - // as nested. - val byName = buckets.find(b => b.identifier.name.nonEmpty && b.identifier.name == leaf.name) - byName.orElse(leaf.bucket.flatMap(attached => buckets.find(_.name == attached.name))) + // Story IDENT-4 -- the decision `bucketFor` recorded FIRST (the level the leaf was attached + // to, found by `Bucket.sameKeyAs`), then a level with the leaf's name AND correlation. It used + // to ask the NAME first, which answered the FIRST of two same-named levels -- a self-join's + // `o.category` / `p.category`, or `YEAR(d)` / `MONTH(d)` -- for either leaf. + // + // 🔴 Resolve through THIS statement's bucket list, never through `Identifier.bucket` itself, + // which is a COPY attached during `update()` and deliberately NOT re-updated (the story-21.3 + // / #253 desync class) -- MEASURED reporting `nestedElement = None` for a bucket the + // statement lists as nested. + def named(b: Bucket): Boolean = b.identifier.name.nonEmpty && b.identifier.name == leaf.name + leaf.bucket + .flatMap(attached => buckets.find(_.sameKeyAs(attached))) + .orElse( + buckets.find(b => + named(b) && + b.identifier.tableAlias.getOrElse(from.mainTableCorrelation) == + leaf.tableAlias.getOrElse(from.mainTableCorrelation) + ) + ) + // ⚠️ The ONE exception, here so that every reader sees it: an ARGUMENT of a NAMELESS + // GROUP BY key (`UPPER(b.category)`, `CONCAT(a.category, b.category)`) keeps the + // correlation-free name match it always had. Such a key has no identity to attach its + // arguments to, so without it `GROUP BY a.category, UPPER(b.category) HAVING + // UPPER(b.category) = 'X'` -- accepted until now and evaluated after the join -- would be + // refused as reading neither a key nor an aggregate. + .orElse(if (namelessKeyArguments.exists(_ eq leaf)) buckets.find(named) else None) + } + + /** The column ARGUMENTS of every HAVING expression that is itself a NAMELESS GROUP BY key -- + * one attached to a GROUP BY level -- the instances [[namedLeavesOf]] hands out (`eq`). + */ + private[this] lazy val namelessKeyArguments: Seq[Identifier] = { + val keys = groupBy.map(_.buckets).getOrElse(Nil) + having + .flatMap(_.criteria) + .toSeq + .flatMap(_.referencedIdentifiers) + .filter(c => c.name.isEmpty && c.bucket.exists(a => keys.exists(_.sameKeyAs(a)))) + .flatMap(c => FunctionUtils.funIdentifiers(c)) + .filter(_.name.nonEmpty) } - private[query] def namedLeavesOf(e: Expression): Seq[Identifier] = + /** Every named column a HAVING expression reads, once each. + * + * Story IDENT-4 -- deduplicated by `(param, tableAlias)`, spelled out: two copies of one + * column through two correlations (`o.category`, `p.category`) are TWO leaves, and + * [[keyBucketOf]] answers them differently, so a dedup that collapsed them would decide a + * verdict. + */ + private[query] def namedLeavesOf(e: Expression): Seq[Identifier] = { + val seen = scala.collection.mutable.HashSet.empty[(String, Option[String])] e.referencedIdentifiers .flatMap(id => FunctionUtils.funIdentifiers(id)) .filter(_.name.nonEmpty) - .distinct + .filter(id => seen.add((id.param, id.tableAlias))) + } /** Why a HAVING leaf that QUALIFIES a SELECT-list alias with a table alias (`HAVING * d.avg_salary > 1` over `AVG(e.salary) AS avg_salary`) cannot be filtered on, or `None` when @@ -1423,18 +1640,51 @@ package object query { } private[query] def keyPredicateOutcome(e: Expression): KeyPredicateOutcome = - if (keyExpressibleByTerms(e)) KeyPredicateOutcome.TermsFilter - else if (readsNestedKey(e)) - KeyPredicateOutcome.Refused( - s"HAVING cannot filter on ${e.sql}: the grouping key is a nested object, and a bucket " + - "filter over it is not expressible. Filter it in WHERE, or group by a field of the parent." - ) - else - KeyPredicateOutcome.Refused( - s"HAVING cannot filter on ${e.sql}: a condition over the GROUP BY key is applied by the " + - "terms filter, which can only express a direct comparison of the key (=, <>, LIKE, IN). " + - "Compare the key itself, or filter in WHERE." - ) + dateHistogramLevelOf(e) match { + case Some(level) => KeyPredicateOutcome.Refused(dateHistogramReason(e, level)) + case None => + if (keyExpressibleByTerms(e)) KeyPredicateOutcome.TermsFilter + else if (readsNestedKey(e)) + KeyPredicateOutcome.Refused( + s"HAVING cannot filter on ${e.sql}: the grouping key is a nested object, and a bucket " + + "filter over it is not expressible. Filter it in WHERE, or group by a field of the parent." + ) + else + KeyPredicateOutcome.Refused( + s"HAVING cannot filter on ${e.sql}: a condition over the GROUP BY key is applied by the " + + "terms filter, which can only express a direct comparison of the key (=, <>, LIKE, IN). " + + "Compare the key itself, or filter in WHERE." + ) + } + + /** Story IDENT-4 -- the level a key predicate addresses when that level is a `date_histogram`: + * asked of the REAL `includes` / `excludes` (as [[keyExpressibleByTerms]] is) and of + * [[Bucket.isDateHistogram]] (as both bridges are). A key whose function returns a date + * (`DATE_TRUNC(d, MONTH)`) is one with no mapping, so the predicate is refused at parse. + * + * Structural, not a type rule: the level the bridges EMIT for such a key is a date histogram, + * whatever the column's mapping turns out to be, and a date histogram has no include / exclude + * -- the key filter has no channel. + * + * Read only through rule (b), which does not run in a statement the RELATIONAL engine + * evaluates ([[relationalClosureRequired]]): no date histogram is built there. + */ + private[query] def dateHistogramLevelOf(e: Expression): Option[Bucket] = { + val empty = BucketIncludesExcludes() + buckets.find(b => + b.isDateHistogram && + (e.includes(b, not = false, empty) != empty || e.excludes(b, not = false, empty) != empty) + ) + } + + /** The stable phrase leads: `GatewayApi.run` relays a parse refusal as its first 120 + * characters, an ellipsis and its last 77. + */ + private[query] def dateHistogramReason(e: Expression, level: Bucket): String = + s"HAVING cannot filter a DATE grouping level: ${e.sql} addresses the GROUP BY key " + + s"${level.sql}, which Elasticsearch groups with a date histogram, and a date histogram has " + + "no key filter. Restrict the rows in WHERE instead (for one month of a date column d: " + + "WHERE d >= '2024-01-01' AND d < '2024-02-01')." /** The first `HAVING` combination whose halves cannot BOTH reach the `terms` filter, or `None` * when every combination is a union the channel expresses. See rule (b2) in `validate()`. @@ -1608,8 +1858,12 @@ package object query { def leaves(c: Criteria): Seq[Expression] = c match { case Predicate(l, _, r, _, _) => leaves(l) ++ leaves(r) case _: ElasticRelation => Nil - case e: Expression => Seq(e).filterNot(_.nested) - case _ => Nil + // Story IDENT-4 -- a MATCH is a leaf too: `Criteria.includes` descends it (its `criteria` + // is the OR of one `MatchCriteria` per column), so every rule must see what the key filter + // applies. + case m: MultiMatchCriteria => leaves(m.criteria) + case e: Expression => Seq(e).filterNot(_.nested) + case _ => Nil } having.flatMap(_.criteria).toSeq.flatMap(leaves) } @@ -1622,6 +1876,7 @@ package object query { private[query] def leavesUnder(c: Criteria): Seq[Expression] = c match { case Predicate(l, _, r, _, _) => leavesUnder(l) ++ leavesUnder(r) case relation: ElasticRelation => leavesUnder(relation.criteria) + case m: MultiMatchCriteria => leavesUnder(m.criteria) case e: Expression => Seq(e) case _ => Nil } @@ -2142,8 +2397,14 @@ package object query { _ <- { // (b) ... and a key predicate neither bucket-level mechanism can express is refused by // the ONE classifier both emission paths read. + // + // Story IDENT-4 -- NOT in a statement the RELATIONAL engine evaluates: no terms filter is + // built there, and the engine evaluates the condition after the join. havingLeaves - .filter(e => groupBy.isDefined && havingScopeOf(e) == HavingScope.GroupKey) + .filter(e => + groupBy.isDefined && !relationalClosureRequired && + havingScopeOf(e) == HavingScope.GroupKey + ) .map(keyPredicateOutcome) .collectFirst { case KeyPredicateOutcome.Refused(reason) => Left(reason) } .getOrElse(Right(())) @@ -2174,7 +2435,10 @@ package object query { // written for the include sense only and applied by a method that is also the exclude // sense -- so it inverted every `<>`. A rule about these two channels that consults only // one of them is wrong by construction. - keyChannelConflict.map(Left(_)).getOrElse(Right(())) + // + // Story IDENT-4 -- not in a statement the RELATIONAL engine evaluates, as rule (b). + if (relationalClosureRequired) Right(()) + else keyChannelConflict.map(Left(_)).getOrElse(Right(())) } _ <- { // (c) 🔴 An OR is honoured by ONE mechanism or by none: a `bucket_selector` script cannot @@ -2223,9 +2487,16 @@ package object query { // symmetry rule is worse than no guard, so the rule is symmetric by construction // instead. A leaf that is not a GROUP BY key still has no level, which is what the // second half says. + // + // Story IDENT-4 -- a level is its GROUP BY rendering (`o.category`, `YEAR(d)`, or the + // alias of a substituted bucket), not its display name: a self-join's two levels, and two + // functions of one column, share a display name and were one "stage", so an OR across + // them -- which Elasticsearch NESTS, i.e. runs as an AND -- was accepted. And no level at + // all in a statement the RELATIONAL engine evaluates: nothing nests there, and the engine + // evaluates the OR after the join -- the level half exists because Elasticsearch nests. def levelOf(e: Expression): String = - if (havingScopeOf(e) != HavingScope.GroupKey) "" - else namedLeavesOf(e).flatMap(keyBucketOf).map(_.name).distinct.sorted.mkString(",") + if (relationalClosureRequired || havingScopeOf(e) != HavingScope.GroupKey) "" + else namedLeavesOf(e).flatMap(keyBucketOf).map(_.sql).distinct.sorted.mkString(",") def firstMixedOr(c: Criteria): Option[(Criteria, Seq[String], Boolean)] = c match { case p @ Predicate(l, op, r, _, _) => val here = @@ -2245,7 +2516,11 @@ package object query { } else None here.orElse(firstMixedOr(l)).orElse(firstMixedOr(r)) case relation: ElasticRelation => firstMixedOr(relation.criteria) - case _ => None + case m: MultiMatchCriteria => + firstMixedOr(m.criteria).map { case (_, names, mechanisms) => + (m, names, mechanisms) + } + case _ => None } having.flatMap(_.criteria).filter(_ => groupBy.isDefined).flatMap(firstMixedOr) match { case Some((p, names, differentMechanisms)) => diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/query/BucketIncludesSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/query/BucketIncludesSpec.scala index eb80c090..e9066731 100644 --- a/sql/src/test/scala/app/softnetwork/elastic/sql/query/BucketIncludesSpec.scala +++ b/sql/src/test/scala/app/softnetwork/elastic/sql/query/BucketIncludesSpec.scala @@ -216,4 +216,56 @@ class BucketIncludesSpec extends AnyFlatSpec with Matchers { it should "leave a statement with no mapping alone -- its key's type is unknown (CHARACTERIZATION)" in { Parser(keyed("n", "n = 1.5")).isRight shouldBe true } + + /** Story IDENT-4 -- the query venue's verdict once the mapping is attached (`resolveWithSchema`): + * IDENT-3's [[verdicts]], which must agree at both resolutions. + */ + private def resolvedVerdict(sql: String): Either[String, Unit] = + verdicts(sql).map(_._2).distinct match { + case Seq(verdict) => verdict + case several => Left(s"the two resolutions disagree: $several") + } + + "a QUALIFIED integer key compared with a fractional number (IDENT-4)" should + "be refused once the mapping is known, in every spelling, and left alone before it" in { + // R1 has no mapping (every column is ANY), so the fractional-key check cannot fire there -- the boundary IDENT-3 + // states, inherited by the qualified spelling. R2 is `SearchApi.resolveWithSchema`. + val refusedAfterResolution = Seq( + "SELECT tb.n, COUNT(*) AS c FROM t tb GROUP BY tb.n HAVING tb.n = 1.5", + "SELECT tb.n, COUNT(*) AS c FROM t tb GROUP BY tb.n HAVING n = 1.5", + "SELECT n, COUNT(*) AS c FROM t tb GROUP BY n HAVING tb.n = 1.5", + "SELECT tb.n, COUNT(*) AS c FROM t tb GROUP BY tb.n HAVING tb.n IN (1.5, 2.5)", + "SELECT tb.n AS k, COUNT(*) AS c FROM t tb GROUP BY k HAVING k = 1.5" + ) + val wrong = refusedAfterResolution.flatMap { sql => + val atParse = Parser(sql) match { + case Right(_: SingleSearch) => None + case other => Some(s"[$sql] R1: expected accepted (no mapping), got $other") + } + val afterResolution = + if (atParse.isDefined) None // reported by atParse + else + resolvedVerdict(sql) match { + case Left(reason) + if reason.contains("HAVING cannot compare the integer GROUP BY key") => + None + case other => Some(s"[$sql] R2: expected the fractional-key refusal, got $other") + } + atParse.toSeq ++ afterResolution.toSeq + } + withClue(s"${wrong.size} wrong:\n${wrong.mkString("\n")}\n") { wrong shouldBe empty } + } + + it should "accept a whole number with a decimal point in every spelling" in { + Seq( + "SELECT tb.n, COUNT(*) AS c FROM t tb GROUP BY tb.n HAVING tb.n = 2.0", + "SELECT tb.n, COUNT(*) AS c FROM t tb GROUP BY tb.n HAVING n = 2.0", + "SELECT n, COUNT(*) AS c FROM t tb GROUP BY n HAVING tb.n = 2.0" + ).flatMap { sql => + resolvedVerdict(sql) match { + case Right(()) => None + case other => Some(s"[$sql] $other") + } + } shouldBe empty + } } diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/query/HavingNullAwareSelectorSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/query/HavingNullAwareSelectorSpec.scala index a9d77879..bc107102 100644 --- a/sql/src/test/scala/app/softnetwork/elastic/sql/query/HavingNullAwareSelectorSpec.scala +++ b/sql/src/test/scala/app/softnetwork/elastic/sql/query/HavingNullAwareSelectorSpec.scala @@ -296,7 +296,17 @@ class HavingNullAwareSelectorSpec extends AnyFlatSpec with Matchers with OptionV it should "leave a MATCH over the GROUP BY key as it is" in { parsed(group + "MATCH (g) AGAINST ('x')") - parsed("SELECT g, COUNT(*) AS c FROM t GROUP BY g, h HAVING MATCH (g, h) AGAINST ('x')") + } + + it should "refuse a MATCH over two GROUP BY keys by the OR rule" in { + // Story IDENT-4 -- over TWO keys a MATCH is an OR across two grouping levels, which + // Elasticsearch nests (it ran as a conjunction): refused by that rule, never by this one. + Parser("SELECT g, COUNT(*) AS c FROM t GROUP BY g, h HAVING MATCH (g, h) AGAINST ('x')") match { + case Left(error) => + error.msg should include("DIFFERENT GROUP BY keys") + error.msg should not include "HAVING cannot apply MATCH" + case Right(other) => fail(s"accepted: $other") + } } it should "leave a MATCH the nested filter applies as it is" in { diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/query/QualifiedGroupKeySpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/query/QualifiedGroupKeySpec.scala new file mode 100644 index 00000000..d550ca69 --- /dev/null +++ b/sql/src/test/scala/app/softnetwork/elastic/sql/query/QualifiedGroupKeySpec.scala @@ -0,0 +1,818 @@ +/* + * 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.{GenericIdentifier, Identifier} +import app.softnetwork.elastic.sql.parser.Parser +import app.softnetwork.elastic.sql.schema.{Column, Table => SchemaTable} +import app.softnetwork.elastic.sql.`type`.SQLTypes +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import scala.util.Try + +/** Story IDENT-4 -- a QUALIFIED GROUP BY key resolves to its bucket in HAVING. + * + * "Which bucket does this leaf name" keyed on a DISPLAY name collapses: after ONE resolution a + * self-join's two levels share a name, and a lookup by name answers the LAST of them for either + * leaf. So every assertion here is made at BOTH resolutions, on a DERIVED population, and the + * self-join is asserted in BOTH directions (asserting only the last level is exactly what a + * collapsed lookup gets right). + */ +class QualifiedGroupKeySpec extends AnyFlatSpec with Matchers { + + import QualifiedKeyPopulation._ + + private val none = BucketIncludesExcludes() + private val nothing = (none, none) + private def keeps(values: String*) = (BucketIncludesExcludes(values = values.toSet), none) + + private def verdict(sql: String): Either[String, Unit] = + Parser(sql) match { + case Right(_) => Right(()) + case Left(err) => Left(err.msg) + } + + /** The HAVING rule a refusal comes from, read off its message (reporting only). */ + private def refusalFamily(reason: String): String = + if (reason.contains("can only filter on a GROUP BY key or on an aggregate")) "(a)" + else if (reason.contains("which can only express a direct comparison of the key")) "(b)" + else if (reason.contains("the grouping key is a nested object")) "(b-nested)" + else if (reason.contains("HAVING cannot combine the conditions")) "(b2)" + else if (reason.contains("DIFFERENT GROUP BY keys")) "(c)-keys" + else if (reason.contains("different mechanisms")) "(c)-mechanisms" + else if (reason.contains("HAVING cannot apply MATCH")) "match-refusal" + else if (reason.contains("MATERIALIZED VIEW cannot")) "view" + else if (reason.contains("HAVING cannot filter a DATE grouping level")) "date-level" + else if (reason.contains("HAVING cannot compare the integer GROUP BY key")) "fractional-key" + else s"other(${reason.take(80)})" + + private def verdictFamily(sql: String): String = + verdict(sql).fold(refusalFamily, _ => "accepted") + + private def channels(ss: SingleSearch): Seq[(BucketIncludesExcludes, BucketIncludesExcludes)] = + channelsOf(ss).map { case (_, kept, removed) => (kept, removed) } + + /** One observation, COLLECTED: `None` when it holds, the observed value otherwise -- a test that + * read as a per-cell map must never stop at its first red (`foreach { … shouldBe … }` does). + */ + private def check(label: String)(actual: => Any, expected: Any): Option[String] = + Try(actual).toEither match { + case Right(a) if a == expected => None + case Right(a) => Some(s"$label: $a, expected $expected") + case Left(t) => Some(s"$label threw $t") + } + + private val dated: SchemaTable = SchemaTable( + "t", + columns = List( + Column("d", SQLTypes.Date), + Column("s", SQLTypes.Keyword), + Column("id", SQLTypes.Keyword), + Column("amount", SQLTypes.Int) + ) + ) + + private val selfJoin = + "SELECT COUNT(*) AS c FROM orders o JOIN orders p ON o.did = p.id GROUP BY o.category, p.category " + + private val selfJoinSchemas = Map("orders" -> ordersSchema) + + // 1 ------------------------------------------------------------------------------------------ + "IDENT-4: the qualified-key population" should + "put each predicate on exactly the level it names, at BOTH resolutions, or refuse it" in { + cells should have size ExpectedCellCount.toLong + val outcomes = cells.map { cell => + val atR1 = Try(r1(cell.sql)) + val atR2 = atR1.flatMap(s => Try(r2(cell.shape, s))) + val problems: Seq[String] = (atR1.map(channels).toOption, atR2.map(channels).toOption) match { + case (Some(c1), Some(c2)) => + Seq( + if (c1 != cell.expectedChannels) Some(s"R1 channels $c1") else None, + if (c2 != cell.expectedChannels) Some(s"R2 channels $c2") else None + ).flatten ++ ( + (verdict(cell.sql), cell.addressed) match { + case (Right(()), Some(_)) => Nil + // a non-addressed cell is refused by its operator's own family: rule (a) for the seven + // comparison operators, the MATCH rule (a MATCH over a column no level names) for MATCH + case (Left(r), None) + if refusalFamily(r) == (if (cell.op.label == "MATCH") "match-refusal" + else "(a)") => + Nil + case (other, expected) => + Seq(s"verdict $other, expected ${if (expected.isDefined) "accepted" + else "refused (neither)"}") + } + ) + case _ => Seq(s"threw: R1 $atR1, R2 $atR2") + } + cell -> problems + } + val wrong = outcomes.filter(_._2.nonEmpty) + // The ORACLE's split, printed for the reader -- a prediction about the alphabet, never asserted. + info( + s"${cells.count(_.addressed.isDefined)} accepted · ${cells.count(_.addressed.isEmpty)} refused · " + + s"${wrong.size} wrong" + ) + // The MEASURED verdicts of the tree under test, by operator kind and refusal family -- printed, + // never asserted. + cells + .groupBy(c => if (c.op.label == "MATCH") "MATCH" else "non-MATCH") + .toSeq + .sortBy(_._1) + .foreach { case (kind, group) => + val split = + group.groupBy(c => verdictFamily(c.sql)).map { case (f, cs) => s"$f=${cs.size}" } + info(s"verdicts, $kind (${group.size} cells): ${split.toSeq.sorted.mkString(" · ")}") + } + withClue( + s"${wrong.size} of ${cells.size} cells wrong:\n" + + wrong + .map { case (c, p) => s"[${c.label}]\n ${c.sql}\n ${p.mkString("; ")}" } + .mkString("\n") + + "\n" + ) { wrong shouldBe empty } + } + + // 2 ------------------------------------------------------------------------------------------ + "a self-join's key filter on one level" should + "sit on EXACTLY the level it names -- in BOTH directions" in { + // Two failures to exclude: the include on EVERY level, and a lookup that answers the LAST + // level -- which gets `p` right and `o` wrong. Hence both directions. + channels(r1(selfJoin + "HAVING p.category = 'x'")) shouldBe Seq(nothing, keeps("x")) + channels(r1(selfJoin + "HAVING o.category = 'x'")) shouldBe Seq(keeps("x"), nothing) + verdict(selfJoin + "HAVING p.category = 'x'") shouldBe Right(()) + verdict(selfJoin + "HAVING o.category = 'x'") shouldBe Right(()) + } + + // 3 ------------------------------------------------------------------------------------------ + "a self-join's key filter after re-resolution" should + "give IDENTICAL channels after a second and a third resolution (updateAll)" in { + Seq("o", "p").foreach { alias => + val once = r1(selfJoin + s"HAVING $alias.category = 'x'") + val twice = once.updateAll(selfJoinSchemas) + withClue(s"[$alias, second] ") { channels(twice) shouldBe channels(once) } + // `Table.mergeWithSearch` re-updates a statement that was already resolved + withClue(s"[$alias, third] ") { + channels(twice.updateAll(selfJoinSchemas)) shouldBe channels(once) + } + } + } + + // 4 ------------------------------------------------------------------------------------------ + "a GROUP BY over a qualified column at both venues" in { + // Accepted and answered before this story at a venue that resolves twice (at a one-resolution + // venue its body lacked the metrics). + // Executable in core, so it is held at RUN, not only at RENDER. + val wrong = qualifiedVenues.flatMap { venue => + val sql = + s"SELECT g.category AS k, COUNT(*) AS c, MAX(amount) AS mx, SUM(amount) AS s FROM $venue " + + "GROUP BY k HAVING k = 'cat_37'" + maxSumRows(sql) match { + case Right(actual) if actual == expectedGroups(_ == "cat_37") => None + case other => Some(s"[$sql] $other") + } + } + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + it should "select exactly the groups it names inside a WHERE subquery" in { + maxSumRows( + "SELECT category AS k, COUNT(*) AS c, MAX(amount) AS mx, SUM(amount) AS s FROM group_by_completeness " + + "WHERE category IN (SELECT g.category FROM group_by_completeness g GROUP BY g.category " + + "HAVING g.category = 'cat_37') GROUP BY category" + ) shouldBe Right(expectedGroups(_ == "cat_37")) + } + + it should "refuse a table's own name once aliased, and an undeclared qualifier -- neither is a key" in { + Seq("group_by_completeness.category", "zz.category").foreach { operand => + val sql = + "SELECT g.category AS k, COUNT(*) AS c FROM group_by_completeness g GROUP BY g.category " + + s"HAVING $operand = 'cat_37'" + withClue(s"[$sql] ") { refusalOf(sql) should include("can only filter on a GROUP BY key") } + } + } + + it should "carry Elasticsearch 6's refusal set on the qualified spelling" in { + // IDENT-3 contract item 4: a pattern in one list beside two or more values in the other. + val sql = + "SELECT g.category AS k, COUNT(*) AS c, MAX(amount) AS mx, SUM(amount) AS s " + + "FROM group_by_completeness g GROUP BY g.category " + + "HAVING g.category LIKE 'cat_3%' AND g.category NOT IN ('cat_30', 'cat_31')" + if (esMajor == 6) refusalOf(sql) should include("on Elasticsearch 6") + else + maxSumRows(sql) shouldBe Right( + expectedGroups(k => k.startsWith("cat_3") && !Set("cat_30", "cat_31")(k)) + ) + } + + it should "refuse a QUALIFIED integer key compared with a fractional number, like the bare one" in { + keyFilterLoaded + val wrong = Seq( + s"SELECT k.lval AS l, COUNT(*) AS c FROM $keyIndex k GROUP BY k.lval HAVING k.lval = 1.5", + s"SELECT k.lval AS l, COUNT(*) AS c FROM $keyIndex k GROUP BY k.lval HAVING lval = 1.5", + s"SELECT lval AS l, COUNT(*) AS c FROM $keyIndex k GROUP BY lval HAVING k.lval = 1.5" + ).flatMap { sql => + Await.result(client.run(sql), 60.seconds) match { + case ElasticFailure(error) + if error.statusCode.contains(400) && + // the refusal names the key as the condition spells it (`lval` or `k.lval`) + error.message.contains("HAVING cannot compare the integer GROUP BY key") => + None + case other => + Some(s"[$sql] must be refused, got $other -- the groups: ${countsBy(sql, "l")}") + } + } + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + countsBy( + s"SELECT k.lval AS l, COUNT(*) AS c FROM $keyIndex k GROUP BY k.lval HAVING k.lval = 2.0", + "l" + ) shouldBe + Right(expectedCounts("lval", _ == "2")) + } + + // ---- the date fixture: two functions of one column, and the date-histogram suspect ---------- + + private val datesIndex = "group_by_qualified_dates" + + /** 2024-01: 1 + 2 documents, 2024-02: 3, 2025-01: 4 -- amounts 1..10 in that order. */ + private val datedDocs: Seq[(String, Int)] = Seq( + "2024-01-03" -> 1, + "2024-01-20" -> 2, + "2024-01-20" -> 3, + "2024-02-03" -> 4, + "2024-02-03" -> 5, + "2024-02-03" -> 6, + "2025-01-05" -> 7, + "2025-01-05" -> 8, + "2025-01-05" -> 9, + "2025-01-05" -> 10 + ) + + private lazy val qualifiedDatesLoaded: Unit = indexDatesFixture(datesIndex) + + private def indexDatesFixture(target: String): Unit = { + val settings = """{"number_of_shards": 3, "number_of_replicas": 0}""" + val mapping = + """{ + | "properties": { + | "id": { "type": "keyword" }, + | "d": { "type": "date" }, + | "amount": { "type": "integer" } + | } + |}""".stripMargin + client.createIndex(target, settings = settings).get shouldBe true + client.setMapping(target, mapping).get shouldBe true + implicit val datesBulkOptions: BulkOptions = BulkOptions(defaultIndex = target, logEvery = 1000) + implicit def datesListToSource[T](list: List[T]): Source[T, NotUsed] = + Source.fromIterator(() => list.iterator) + val docs = datedDocs.zipWithIndex.map { case ((d, amount), i) => + s"""{"id":"q$i","d":"$d","amount":$amount}""" + }.toList + client.bulk[String](docs, identity, idKey = Some(Set("id"))) match { + case ElasticSuccess(_) => + case ElasticFailure(error) => + error.cause.foreach(_.printStackTrace()) + fail(s"Bulk indexing $target failed: ${error.message}") + } + client.refresh(target) + } + + "IDENT-4: two functions of one date column as two keys" should + "REFUSE an OR across the two levels, which Elasticsearch would run as an AND" in { + qualifiedDatesLoaded + // Answered until now, with the conjunction `[(3,3,6)]` at a one-resolution venue. + val wrong = Seq( + s"SELECT COUNT(*) AS c, MAX(amount) AS mx, SUM(amount) AS s FROM $datesIndex " + + "GROUP BY YEAR(d), MONTH(d) HAVING YEAR(d) = 2024 OR MONTH(d) = 1", + s"SELECT COUNT(*) AS c, MAX(amount) AS mx, SUM(amount) AS s FROM $datesIndex q " + + "GROUP BY YEAR(q.d), MONTH(q.d) HAVING YEAR(q.d) = 2024 OR MONTH(q.d) = 1" + ).flatMap { sql => + Await.result(client.run(sql), 60.seconds) match { + case ElasticFailure(error) if error.message.contains("DIFFERENT GROUP BY keys") => None + case _ => Some(s"[$sql] must be refused, got ${metricTriples(sql)}") + } + } + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + // The date-histogram key-filter refusal (required). + "IDENT-4: a key predicate on a DATE grouping level" should + "be refused by name -- and the remedy the refusal names must answer" in { + // Answered until now with the unfiltered groups: the date histogram dropped the condition, at + // every venue. + qualifiedDatesLoaded + val wrong = Seq( + s"SELECT COUNT(*) AS c FROM $datesIndex GROUP BY DATE_TRUNC(d, MONTH) " + + "HAVING DATE_TRUNC(d, MONTH) = '2024-01-01'", + s"SELECT COUNT(*) AS c FROM $datesIndex q GROUP BY DATE_TRUNC(q.d, MONTH) " + + "HAVING DATE_TRUNC(q.d, MONTH) = '2024-01-01'", + s"SELECT COUNT(*) AS c FROM $datesPattern q GROUP BY DATE_TRUNC(q.d, MONTH) " + + "HAVING DATE_TRUNC(q.d, MONTH) = '2024-01-01'" + ).flatMap { sql => + Await.result(client.run(sql), 60.seconds) match { + case ElasticFailure(error) + if error.message.contains("HAVING cannot filter a DATE grouping level") => + None + case _ => Some(s"[$sql] must be refused, got ${dateGroups(sql)}") + } + } + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + metricTriples( + s"SELECT COUNT(*) AS c, MAX(amount) AS mx, SUM(amount) AS s FROM $datesIndex q " + + "WHERE q.d >= '2024-01-01' AND q.d < '2024-02-01' GROUP BY DATE_TRUNC(q.d, MONTH)" + ) shouldBe Right(Seq((3L, 3L, 6L))) + } + + // The HAVING rules seeing a MATCH (required). + "IDENT-4: a MATCH on the GROUP BY key" should "be refused where the rules cannot honour it" in { + // COLLECTED, both venues and both spellings: answered until now with the intersection + // `{cat_37}` where SQL means `{cat_36, cat_37}`. + val combinations = for { + venue <- qualifiedVenues + (select, key) <- Seq("category" -> "category", "g.category" -> "g.category") + } yield s"SELECT $select AS k, COUNT(*) AS c FROM $venue GROUP BY $key " + + s"HAVING MATCH ($key) AGAINST ('cat_37') OR COUNT(*) > 35" + // The UNALIASED form too (metric-only, groups told apart by their triples). + val unaliased = for { + venue <- qualifiedVenues + key <- Seq("category", "g.category") + } yield s"SELECT COUNT(*) AS c, MAX(amount) AS mx, SUM(amount) AS s FROM $venue GROUP BY $key " + + s"HAVING MATCH ($key) AGAINST ('cat_37') OR COUNT(*) > 35" + val wrong = ((combinations ++ unaliased).map(_ -> "different mechanisms") :+ ( + // since #408 a MATCH over a column that is neither a key nor an aggregate is refused by the + // MATCH rule itself (not a witness of the HAVING rules seeing a MATCH) + "SELECT category AS k, COUNT(*) AS c FROM group_by_completeness GROUP BY category " + + "HAVING MATCH (id) AGAINST ('cat_37_1')" -> "HAVING cannot apply MATCH" + )).flatMap { case (sql, reason) => + Await.result(client.run(sql), 60.seconds) match { + case ElasticFailure(error) if error.message.contains(reason) => None + case _ => + val answered = + if (sql.contains(" AS k,")) maxSumRows(sql).map(_.keys.toSeq.sorted) + else metricTriples(sql) + Some(s"[$sql] must be refused ($reason), got $answered") + } + } + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + // ---- shared rows of the date fixture --------------------------------------------------------- + + private val datesPattern = "group_by_qualified_d*" + + /** The document count of every group, sorted -- each date and each month of the date fixture has + * its own count, so the counts name the groups. + */ + private def dateGroups(sql: String): Either[String, Seq[Long]] = + metricTriples(sql).map(_.map(_._1)) + + // ---- a temporal WHERE comparison of a GROUP BY key column ---------------------------------- + // + // Two date fields; the right operand of `q.d > DATE_ADD(q.e, INTERVAL 1 DAY)` is temporal without + // a mapping, so the comparison is a date one at every resolution. SQL: `d > e + 1 day` keeps r1 + // only. Run at the one-resolution venue below. + + private def indexIm10Fixture(target: String): Unit = { + val settings = """{"number_of_shards": 1, "number_of_replicas": 0}""" + val mapping = + """{"properties": {"id": {"type": "keyword"}, "d": {"type": "date"}, "e": {"type": "date"}}}""" + client.createIndex(target, settings = settings).get shouldBe true + client.setMapping(target, mapping).get shouldBe true + implicit val im10BulkOptions: BulkOptions = BulkOptions(defaultIndex = target, logEvery = 1000) + implicit def im10ListToSource[T](list: List[T]): Source[T, NotUsed] = + Source.fromIterator(() => list.iterator) + val docs = List( + """{"id":"r1","d":"2024-01-10","e":"2024-01-01"}""", + """{"id":"r2","d":"2024-01-02","e":"2024-01-01"}""", + """{"id":"r3","d":"2024-01-01","e":"2024-01-05"}""" + ) + client.bulk[String](docs, identity, idKey = Some(Set("id"))) match { + case ElasticSuccess(_) => client.refresh(target) + case ElasticFailure(error) => + error.cause.foreach(_.printStackTrace()) + fail(s"Bulk indexing $target failed: ${error.message}") + } + } + + private def im10Rows(index: String): Seq[(String, Seq[Long])] = Seq( + s"SELECT COUNT(*) AS c FROM $index q WHERE q.d > DATE_ADD(q.e, INTERVAL 1 DAY) GROUP BY q.d" -> Seq( + 1L + ), + s"SELECT COUNT(*) AS c FROM $index WHERE d > DATE_ADD(e, INTERVAL 1 DAY) GROUP BY d" -> Seq(1L) + ) + + // ---- the fractional key check over an index pattern (since #412 a pattern is typed) --------- + + "IDENT-4: an integer key compared with a fractional number over an index pattern" should + "be refused in every spelling" in { + keyFilterLoaded + val wrong = Seq( + "SELECT lval AS l, COUNT(*) AS c FROM group_by_key_filter* k GROUP BY lval HAVING lval = 1.5", + "SELECT k.lval AS l, COUNT(*) AS c FROM group_by_key_filter* k GROUP BY k.lval HAVING k.lval = 1.5" + ).flatMap { sql => + Await.result(client.run(sql), 60.seconds) match { + case ElasticFailure(error) + if error.message.contains("HAVING cannot compare the integer GROUP BY key") => + None + case other => + Some( + s"[$sql] must be refused, got ${other.toString.take(300)} -- groups ${countsBy(sql, "l")}" + ) + } + } + withClue(wrong.mkString("\n")) { wrong shouldBe empty } + } + + // ---- the ONE-resolution venue: a remembered schema miss ------------------------------------- + // + // Every one-update path of `resolveWithSchema` sends the same unresolved statement -- a failed or + // denied mapping read, a remembered 404, a client that is not an `IndicesApi` -- so a FROM queried + // BEFORE its index exists runs, while the miss is remembered (`schema-cache.ttl`, at most five + // minutes), the bytes a user without the mapping privilege runs. Each index below is armed that + // way, then created with the mapping and documents of its twin above; that the rows ran at ONE + // resolution is proven by the MECHANISM, before and after them: the statement comes back out of + // the resolution as the SAME instance (no mapping was attached). + + private val oneResolutionIndex = "group_by_one_resolution" + + private val oneResolutionDates = "group_by_one_resolution_dates" + + private val im10OneResolution = "group_by_im10_one" + + @scala.annotation.tailrec + private def baseClientOf(c: ElasticClientApi): ElasticClientApi = c match { + case delegator: ElasticClientDelegator => baseClientOf(delegator.delegate) + case other => other + } + + /** True when the statement is sent as parsed: no mapping was looked up. */ + private def resolvedOnce(sql: String): Boolean = + Parser(sql) match { + case Right(single: SingleSearch) => + baseClientOf(client).resolveWithSchema(single) match { + case ElasticSuccess(sent) => sent eq single + case ElasticFailure(_) => false + } + case _ => false + } + + private def armSchemaMiss(index: String): Unit = { + implicit val armCtx: ConversionContext = NativeContext + // the index does not exist yet: the lookup answers 404 and the miss is remembered + client.search(SelectStatement(s"SELECT COUNT(*) AS c FROM $index g")) match { + case ElasticSuccess(response) => + fail(s"[$index] exists already: ${response.results.size} rows") + case ElasticFailure(_) => () + } + } + + private def indexCategoriesFixture(target: String): Unit = { + val settings = """{"number_of_shards": 3, "number_of_replicas": 0}""" + val mapping = + """{ + | "properties": { + | "id": { "type": "keyword" }, + | "category": { "type": "keyword" }, + | "amount": { "type": "integer" } + | } + |}""".stripMargin + client.createIndex(target, settings = settings).get shouldBe true + client.setMapping(target, mapping).get shouldBe true + val docs = (for { + c <- 1 to categories + d <- 1 to c + } yield { + val category = f"cat_$c%02d" + s"""{"id":"${category}_$d","category":"$category","amount":$d}""" + }).toList + implicit val categoriesBulkOptions: BulkOptions = + BulkOptions(defaultIndex = target, logEvery = 1000) + implicit def categoriesListToSource[T](list: List[T]): Source[T, NotUsed] = + Source.fromIterator(() => list.iterator) + client.bulk[String](docs, identity, idKey = Some(Set("id"))) match { + case ElasticSuccess(_) => client.refresh(target) + case ElasticFailure(error) => + error.cause.foreach(_.printStackTrace()) + fail(s"Bulk indexing $target failed: ${error.message}") + } + } + + "IDENT-4 at a ONE-resolution venue (a remembered schema miss)" should + "answer every qualified row as the bare spelling does" in { + val armed = Seq(oneResolutionIndex, oneResolutionDates, im10OneResolution) + armed.foreach(armSchemaMiss) + indexCategoriesFixture(oneResolutionIndex) + indexDatesFixture(oneResolutionDates) + indexIm10Fixture(im10OneResolution) + val started = System.currentTimeMillis() + def witnesses: Seq[(String, Boolean)] = + armed.map(i => i -> resolvedOnce(s"SELECT COUNT(*) AS c FROM $i g")) + val before = witnesses + val venue = s"$oneResolutionIndex g" + val dates = s"$oneResolutionDates q" + val metrics = "COUNT(*) AS c, MAX(amount) AS mx, SUM(amount) AS s" + val (matrixSize, matrixWrong) = qualifiedMatrix(Seq(venue)) + def row(sql: String, expected: Any, answered: => Any): Option[String] = { + val got = answered + if (got == expected) None + else Some(s"[$sql] expected $expected, got ${got.toString.take(400)}") + } + def refused(sql: String, reason: String, answered: => Any): Option[String] = + Await.result(client.run(sql), 60.seconds) match { + case ElasticFailure(error) if error.message.contains(reason) => None + case _ => row(sql, s"refused ($reason)", answered) + } + val named = Seq( + // A qualified key's metrics, and a HAVING over them + row( + s"SELECT g.category AS k, $metrics FROM $venue GROUP BY g.category", + Right(expectedGroups(_ => true)), + maxSumRows(s"SELECT g.category AS k, $metrics FROM $venue GROUP BY g.category") + ), + row( + s"SELECT g.category AS k, $metrics FROM $venue GROUP BY g.category HAVING COUNT(*) > 35", + Right(expectedGroups(Set("cat_36", "cat_37"))), + maxSumRows( + s"SELECT g.category AS k, $metrics FROM $venue GROUP BY g.category HAVING COUNT(*) > 35" + ) + ), + // the SELECT-alias remedy + row( + s"SELECT g.category AS k, $metrics FROM $venue GROUP BY g.category HAVING k = 'cat_37'", + Right(expectedGroups(_ == "cat_37")), + maxSumRows( + s"SELECT g.category AS k, $metrics FROM $venue GROUP BY g.category HAVING k = 'cat_37'" + ) + ), + row( + s"SELECT g.category AS k, $metrics FROM $venue GROUP BY k HAVING k = 'cat_37'", + Right(expectedGroups(_ == "cat_37")), + maxSumRows(s"SELECT g.category AS k, $metrics FROM $venue GROUP BY k HAVING k = 'cat_37'") + ) + ) ++ (for { + // A MATCH on the key: aliased and unaliased, bare and qualified + key <- Seq("category", "g.category") + sql <- Seq( + s"SELECT $key AS k, $metrics FROM $venue GROUP BY $key HAVING MATCH ($key) AGAINST ('cat_37') OR COUNT(*) > 35", + s"SELECT $metrics FROM $venue GROUP BY $key HAVING MATCH ($key) AGAINST ('cat_37') OR COUNT(*) > 35" + ) + } yield refused(sql, "different mechanisms", metricTriples(sql))) ++ Seq( + // Two functions of one column keeping their own filters, bare and qualified + row( + s"SELECT $metrics FROM $dates GROUP BY YEAR(d), MONTH(d) HAVING MONTH(d) = 1", + Right(Seq((3L, 3L, 6L), (4L, 10L, 34L))), + metricTriples(s"SELECT $metrics FROM $dates GROUP BY YEAR(d), MONTH(d) HAVING MONTH(d) = 1") + ), + row( + s"SELECT $metrics FROM $dates GROUP BY YEAR(q.d), MONTH(q.d) HAVING MONTH(q.d) = 1", + Right(Seq((3L, 3L, 6L), (4L, 10L, 34L))), + metricTriples( + s"SELECT $metrics FROM $dates GROUP BY YEAR(q.d), MONTH(q.d) HAVING MONTH(q.d) = 1" + ) + ), + // The AND of the two levels, each half on its own level (refused until now) + row( + s"SELECT $metrics FROM $dates GROUP BY YEAR(d), MONTH(d) HAVING YEAR(d) = 2024 AND MONTH(d) = 1", + Right(Seq((3L, 3L, 6L))), + metricTriples( + s"SELECT $metrics FROM $dates GROUP BY YEAR(d), MONTH(d) HAVING YEAR(d) = 2024 AND MONTH(d) = 1" + ) + ), + row( + s"SELECT $metrics FROM $dates GROUP BY YEAR(q.d), MONTH(q.d) HAVING YEAR(q.d) = 2024 AND MONTH(q.d) = 1", + Right(Seq((3L, 3L, 6L))), + metricTriples( + s"SELECT $metrics FROM $dates GROUP BY YEAR(q.d), MONTH(q.d) HAVING YEAR(q.d) = 2024 AND MONTH(q.d) = 1" + ) + ), + // An OR across the two levels, bare and qualified (answered until now as the AND) + refused( + s"SELECT $metrics FROM $dates GROUP BY YEAR(d), MONTH(d) HAVING YEAR(d) = 2024 OR MONTH(d) = 1", + "DIFFERENT GROUP BY keys", + metricTriples( + s"SELECT $metrics FROM $dates GROUP BY YEAR(d), MONTH(d) HAVING YEAR(d) = 2024 OR MONTH(d) = 1" + ) + ), + refused( + s"SELECT $metrics FROM $dates GROUP BY YEAR(q.d), MONTH(q.d) HAVING YEAR(q.d) = 2024 OR MONTH(q.d) = 1", + "DIFFERENT GROUP BY keys", + metricTriples( + s"SELECT $metrics FROM $dates GROUP BY YEAR(q.d), MONTH(q.d) HAVING YEAR(q.d) = 2024 OR MONTH(q.d) = 1" + ) + ), + // A DATE_TRUNC key filter refused at every venue, bare and qualified; a DATE key at + // this venue is a terms level on a date field -- its filter applies (the one date of count 1) + refused( + s"SELECT $metrics FROM $dates GROUP BY DATE_TRUNC(d, MONTH) HAVING DATE_TRUNC(d, MONTH) = '2024-01-01'", + "HAVING cannot filter a DATE grouping level", + metricTriples( + s"SELECT $metrics FROM $dates GROUP BY DATE_TRUNC(d, MONTH) HAVING DATE_TRUNC(d, MONTH) = '2024-01-01'" + ) + ), + refused( + s"SELECT COUNT(*) AS c FROM $dates GROUP BY DATE_TRUNC(q.d, MONTH) HAVING DATE_TRUNC(q.d, MONTH) = '2024-01-01'", + "HAVING cannot filter a DATE grouping level", + dateGroups( + s"SELECT COUNT(*) AS c FROM $dates GROUP BY DATE_TRUNC(q.d, MONTH) HAVING DATE_TRUNC(q.d, MONTH) = '2024-01-01'" + ) + ), + row( + s"SELECT COUNT(*) AS c FROM $dates GROUP BY q.d HAVING q.d = '2024-01-03'", + Right(Seq(1L)), + dateGroups(s"SELECT COUNT(*) AS c FROM $dates GROUP BY q.d HAVING q.d = '2024-01-03'") + ), + row( + s"SELECT COUNT(*) AS c FROM $dates GROUP BY d HAVING d = '2024-01-03'", + Right(Seq(1L)), + dateGroups(s"SELECT COUNT(*) AS c FROM $dates GROUP BY d HAVING d = '2024-01-03'") + ), + row( + s"SELECT COUNT(*) AS c FROM $dates GROUP BY d", + Right(Seq(1L, 2L, 3L, 4L)), + dateGroups(s"SELECT COUNT(*) AS c FROM $dates GROUP BY d") + ) + ) ++ ( + // a temporal WHERE comparison of a GROUP BY key column, qualified or bare (Elasticsearch 7 and + // up, which hand a script a genuine `ZonedDateTime`) + if (esMajor < 7) Nil + else + im10Rows(im10OneResolution).map { case (sql, expected) => + row(sql, Right(expected), dateGroups(sql)) + } + ) + val after = witnesses + val elapsed = System.currentTimeMillis() - started + val wrong = matrixWrong ++ named.flatten + withClue(s"the schema miss did not stand for every row: before $before, after $after") { + (before ++ after).forall(_._2) shouldBe true + } + elapsed should be < SchemaCacheSettings.DefaultTtlMs + withClue(s"${wrong.size} wrong ($matrixSize matrix cells):\n${wrong.mkString("\n")}\n") { + wrong shouldBe empty + } + } }