diff --git a/README.md b/README.md index e9094537..ce6dc434 100644 --- a/README.md +++ b/README.md @@ -162,7 +162,7 @@ Precomputed, automatically refreshed query results stored as Elasticsearch indic ```sql CREATE OR REPLACE MATERIALIZED VIEW orders_with_customers_mv REFRESH EVERY 10 SECONDS -WITH (delay = '2s', user_latency = '1s') +WITH (delay = '1s', user_latency = '1s') AS SELECT o.id, diff --git a/documentation/sql/materialized_views.md b/documentation/sql/materialized_views.md index 01eaf4e7..29208b69 100644 --- a/documentation/sql/materialized_views.md +++ b/documentation/sql/materialized_views.md @@ -91,7 +91,7 @@ AS select_statement The `REFRESH EVERY` clause controls how frequently transforms check for new data. ```sql -REFRESH EVERY 10 SECONDS +REFRESH EVERY 30 SECONDS REFRESH EVERY 5 MINUTES REFRESH EVERY 1 HOUR ``` @@ -111,7 +111,11 @@ WITH (delay = '5s', user_latency = '1s') --- -### Simple Materialized View (no JOIN) +### Single-table Materialized View (no JOIN) + +A materialized view over a **single table is supported**. With no JOIN there is no enrichment chain: +the engine generates exactly **one** transform, reading the source table and writing the view index, +applying the `WHERE`, `GROUP BY` and aggregations of the definition. ```sql CREATE MATERIALIZED VIEW active_orders_mv @@ -122,6 +126,34 @@ FROM orders WHERE status = 'active'; ``` +This creates: +- One transform (source → view) — no changelog transform, no enrich policy, no ingest pipeline +- **No watcher** — a single-table view is therefore the one materialized-view shape that runs on a + **basic** Elasticsearch licence (see [Watcher Dependency and Elasticsearch Licensing](#watcher-dependency-and-elasticsearch-licensing)) +- The view index `active_orders_mv` + +A single-table view whose `SELECT` has no `WHERE`, no `GROUP BY` and no aggregation is also accepted — +it materialises the projected columns of the source table. + +#### Minimum refresh interval + +When `REFRESH EVERY` is given **without** an explicit `delay`, the engine derives the per-transform +delay from the frequency. Every transform must be able to run twice per refresh, so: + +``` +REFRESH EVERY ≥ 2 × (number of transforms) × 10 seconds +``` + +| View shape | Transforms | Minimum `REFRESH EVERY` | +|---|---|---| +| Single table (no JOIN) | 1 | 20 seconds | +| One JOIN + `WHERE` | 3 (changelog + enrichment + final) | 60 seconds | +| One JOIN + `WHERE` + computed columns | 4 (+ computed-fields) | 80 seconds | + +Below that the statement is rejected with *"Calculated delay (N seconds) is too small … Minimum +required frequency: M seconds"*. Supply an explicit `WITH (delay = '…')` to use a shorter frequency — +the delay you give is then used as-is, subject only to `delay × 2 × transforms ≤ frequency`. + --- ### Materialized View with JOIN @@ -129,7 +161,7 @@ WHERE status = 'active'; ```sql CREATE OR REPLACE MATERIALIZED VIEW orders_with_customers_mv REFRESH EVERY 8 SECONDS -WITH (delay = '2s', user_latency = '1s') +WITH (delay = '1s', user_latency = '1s') AS SELECT o.id, @@ -297,7 +329,7 @@ Returns: ```sql CREATE OR REPLACE MATERIALIZED VIEW orders_with_customers_mv REFRESH EVERY 8 SECONDS -WITH (delay = '2s', user_latency = '1s') +WITH (delay = '1s', user_latency = '1s') AS SELECT o.id, o.amount, c.name AS customer_name, c.email, ... FROM orders AS o @@ -393,7 +425,7 @@ COPY INTO customers FROM '/data/customers.json' WITH (format = 'json'); ```sql CREATE OR REPLACE MATERIALIZED VIEW orders_with_customers_mv REFRESH EVERY 8 SECONDS -WITH (delay = '2s', user_latency = '1s') +WITH (delay = '1s', user_latency = '1s') AS SELECT o.id, diff --git a/sql/src/main/scala/app/softnetwork/elastic/sql/transform/TransformTimeUnit.scala b/sql/src/main/scala/app/softnetwork/elastic/sql/transform/TransformTimeUnit.scala index 35cf39b6..7509442a 100644 --- a/sql/src/main/scala/app/softnetwork/elastic/sql/transform/TransformTimeUnit.scala +++ b/sql/src/main/scala/app/softnetwork/elastic/sql/transform/TransformTimeUnit.scala @@ -150,50 +150,62 @@ case class Delay( object Delay { val Default: Delay = Delay(TransformTimeUnit.Minutes, 1) + /** Smallest per-transform delay a calculated refresh chain may use, in seconds. */ + val MinDelaySeconds: Long = 10L + def fromSeconds(seconds: Long): Delay = { val timeInterval = TransformTimeInterval.fromSeconds(seconds) Delay(timeInterval._1, timeInterval._2) } - /** Calculates optimal delay based on frequency and number of stages - * - * Formula: delay = frequency / (nb_stages * buffer_factor) + /** Calculates the optimal delay for one transform of a refresh chain. * - * This ensures the complete chain can refresh within the specified frequency. The buffer factor - * adds safety margin for processing time. + * Hard constraint (identical to the one [[validate]] enforces): every transform runs every + * `delay × 2`, so the whole chain fits inside `frequency` only when `delay ≤ frequency / (2 × + * nbStages)`. `bufferFactor` adds margin on top of that ceiling — it can only ever make the + * delay smaller, never push it past the constraint. With the default `1.5` (indeed with any + * value ≤ 2) the ceiling always wins and the buffered term is inert; it binds only above 2. Keep + * the `min` regardless — it is what absorbs `bufferFactor = 0`, where the buffered term is + * `Infinity.toLong == Long.MaxValue`. * * @param frequency * Desired refresh frequency * @param nbStages - * Total number of stages (changelog + enrichment + aggregate) + * Total number of stages (changelog + enrichment + computed + final). Values below 1 are + * clamped to 1: a materialized view always has at least one transform (the one that writes the + * view index), so a non-positive count is a caller-side accounting bug and must never become + * the user-visible face of a legitimate SQL statement (SoftClient4ES#185). * @param bufferFactor * Safety factor (default 1.5) * @return - * Optimal delay for each Transform, or error if constraints cannot be met + * Optimal delay for each Transform, or an actionable error naming the minimum frequency */ def calculateOptimal( frequency: Frequency, nbStages: Int, bufferFactor: Double = 1.5 ): Either[String, Delay] = { - if (nbStages <= 0) { - return Left("Number of stages must be positive") - } - + val stages = math.max(nbStages, 1) val frequencySeconds = frequency.toSeconds - val optimalDelaySeconds = (frequencySeconds / (nbStages * bufferFactor)).toInt - // Validate constraints - if (optimalDelaySeconds < 10) { + val maxDelaySeconds: Long = frequencySeconds / (2L * stages) + val bufferedDelaySeconds: Long = (frequencySeconds / (stages * bufferFactor)).toLong + val optimalDelaySeconds: Long = math.min(bufferedDelaySeconds, maxDelaySeconds) + + if (optimalDelaySeconds < MinDelaySeconds) { + // The smallest frequency that would actually be accepted: BOTH terms of the `min` above must + // reach MinDelaySeconds, so the ceiling needs `2 × stages × Min` and the buffered term needs + // `stages × bufferFactor × Min` (rounded up, since the division truncates). Quoting only the + // first would print an unreachable figure whenever bufferFactor > 2. + val minFrequencySeconds: Long = + math.max( + 2L * stages * MinDelaySeconds, + math.ceil(stages * bufferFactor * MinDelaySeconds).toLong + ) Left( s"Calculated delay ($optimalDelaySeconds seconds) is too small. " + s"Consider increasing frequency or reducing number of stages. " + - s"Minimum required frequency: ${nbStages * bufferFactor * 10} seconds" - ) - } else if (optimalDelaySeconds > frequencySeconds / 2) { - Left( - s"Calculated delay ($optimalDelaySeconds seconds) is too large. " + - s"Each stage needs at least delay × 2 = frequency." + s"Minimum required frequency: $minFrequencySeconds seconds" ) } else { Right(Delay.fromSeconds(optimalDelaySeconds)) diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/parser/ParserSpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/parser/ParserSpec.scala index 984121b0..414e735c 100644 --- a/sql/src/test/scala/app/softnetwork/elastic/sql/parser/ParserSpec.scala +++ b/sql/src/test/scala/app/softnetwork/elastic/sql/parser/ParserSpec.scala @@ -37,7 +37,7 @@ import app.softnetwork.elastic.sql.schema.{ } import app.softnetwork.elastic.sql.time.TimeUnit.DAYS import app.softnetwork.elastic.sql.time.{CalendarInterval, TimeUnit} -import app.softnetwork.elastic.sql.transform.{Delay, TransformTimeUnit} +import app.softnetwork.elastic.sql.transform.{Delay, Frequency, TransformTimeUnit} import app.softnetwork.elastic.sql.watcher._ import com.fasterxml.jackson.databind.JsonNode import org.scalatest.flatspec.AnyFlatSpec @@ -2038,6 +2038,63 @@ class ParserSpec extends AnyFlatSpec with Matchers { } } + behavior of "Parser DDL with Materialized View Statements" + + // SoftClient4ES#185 — single-table materialized views are SUPPORTED; the parser must keep + // accepting them. Do not "fix" #185 by adding a JOIN requirement here. + it should "parse a single-table CREATE MATERIALIZED VIEW (no JOIN)" in { + val sql = + """CREATE MATERIALIZED VIEW active_orders_mv + |REFRESH EVERY 30 SECONDS + |AS SELECT id, amount, status FROM orders WHERE status = 'active'""".stripMargin + val result = Parser(sql) + result.isRight shouldBe true + result.toOption.get match { + case create: CreateMaterializedView => + create.view shouldBe "active_orders_mv" + create.frequency shouldBe Some(Frequency(TransformTimeUnit.Seconds, 30)) + create.search.from.joinedTables shouldBe empty + create.search.from.enrichmentRequired shouldBe false + case other => fail(s"Expected CreateMaterializedView, got $other") + } + } + + it should "parse a single-table CREATE MATERIALIZED VIEW with neither WHERE nor GROUP BY" in { + val sql = "CREATE MATERIALIZED VIEW orders_copy_mv AS SELECT id, amount FROM orders" + val result = Parser(sql) + result.isRight shouldBe true + result.toOption.get match { + case create: CreateMaterializedView => + create.frequency shouldBe None + create.search.from.enrichmentRequired shouldBe false + case other => fail(s"Expected CreateMaterializedView, got $other") + } + } + + it should "parse CREATE OR REPLACE MATERIALIZED VIEW with a JOIN and options" in { + // 16s / 2s keeps `delay` and `user_latency` distinct (so a parser that swapped them would be + // caught) while remaining a statement the engine accepts: this view is 4 transforms, and + // Delay.validate(2s, 16s, 4) requires 2 x 2 x 4 = 16 <= 16. + val sql = + """CREATE OR REPLACE MATERIALIZED VIEW orders_with_customers_mv + |REFRESH EVERY 16 SECONDS + |WITH (delay = '2s', user_latency = '1s') + |AS SELECT o.id, c.name AS customer_name + |FROM orders AS o JOIN customers AS c ON o.customer_id = c.id + |WHERE o.status = 'completed'""".stripMargin + val result = Parser(sql) + result.isRight shouldBe true + result.toOption.get match { + case create: CreateMaterializedView => + create.orReplace shouldBe true + create.frequency shouldBe Some(Frequency(TransformTimeUnit.Seconds, 16)) + create.delay shouldBe Some(Delay(TransformTimeUnit.Seconds, 2)) + create.search.from.joinedTables should contain only "customers" + create.search.from.enrichmentRequired shouldBe true + case other => fail(s"Expected CreateMaterializedView, got $other") + } + } + behavior of "Parser DDL with Pipeline Statements" it should "parse CREATE OR REPLACE PIPELINE" in { diff --git a/sql/src/test/scala/app/softnetwork/elastic/sql/transform/DelaySpec.scala b/sql/src/test/scala/app/softnetwork/elastic/sql/transform/DelaySpec.scala new file mode 100644 index 00000000..bcfbbc06 --- /dev/null +++ b/sql/src/test/scala/app/softnetwork/elastic/sql/transform/DelaySpec.scala @@ -0,0 +1,131 @@ +/* + * 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.transform + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +class DelaySpec extends AnyFlatSpec with Matchers { + + behavior of "Delay.calculateOptimal" + + // A single-table materialized view is a one-stage chain: source -> view (SoftClient4ES#185). + it should "derive a usable delay for a single-stage chain" in { + val frequency = Frequency(TransformTimeUnit.Seconds, 30) + Delay.calculateOptimal(frequency, nbStages = 1) match { + case Right(delay) => + delay.toSeconds shouldBe 15L + // Whatever it returns MUST satisfy the invariant its sibling enforces. + Delay.validate(delay, frequency, 1) shouldBe Right(()) + case Left(error) => fail(s"Expected a delay, got: $error") + } + } + + it should "clamp a non-positive stage count instead of surfacing an internal invariant" in { + // 60s / (2 * 1) = 30s ; the old code returned Left("Number of stages must be positive"). + Delay.calculateOptimal(Frequency(TransformTimeUnit.Minutes, 1), nbStages = 0) shouldBe + Right(Delay(TransformTimeUnit.Seconds, 30)) + // The scaladoc clamps "values below 1", not just 0. + Delay.calculateOptimal(Frequency(TransformTimeUnit.Minutes, 1), nbStages = -5) shouldBe + Right(Delay(TransformTimeUnit.Seconds, 30)) + } + + // Pins the published minimum-refresh-interval table in documentation/sql/materialized_views.md: + // 1 transform -> 20s, 3 -> 60s, 4 -> 80s. A doc number with no test on either side of it drifts. + it should "accept exactly the documented minimum frequency and reject one second below it" in { + Seq(1 -> 20L, 3 -> 60L, 4 -> 80L).foreach { case (stages, minFrequency) => + withClue(s"stages=$stages at the documented floor ${minFrequency}s: ") { + val accepted = + Delay.calculateOptimal(Frequency(TransformTimeUnit.Seconds, minFrequency), stages) + accepted.map(_.toSeconds) shouldBe Right(Delay.MinDelaySeconds) + accepted.map(d => + Delay.validate(d, Frequency(TransformTimeUnit.Seconds, minFrequency), stages) + ) shouldBe Right(Right(())) + } + withClue(s"stages=$stages one second below the documented floor: ") { + Delay.calculateOptimal( + Frequency(TransformTimeUnit.Seconds, minFrequency - 1), + stages + ) match { + case Left(error) => error should include(s"Minimum required frequency: $minFrequency") + case Right(delay) => fail(s"Expected a rejection below the floor, got: $delay") + } + } + } + } + + // The figure quoted by the rejection must be reachable for a non-default buffer factor too — + // quoting only the `2 × stages × 10` ceiling would name a frequency that still fails. + it should "name a minimum frequency that actually succeeds for a non-default buffer factor" in { + val bufferFactor = 3.0 + Delay.calculateOptimal( + Frequency(TransformTimeUnit.Seconds, 20), + nbStages = 1, + bufferFactor = bufferFactor + ) match { + case Left(error) => + error should include("Minimum required frequency: 30 seconds") + // Taking the advice must work. + Delay + .calculateOptimal( + Frequency(TransformTimeUnit.Seconds, 30), + nbStages = 1, + bufferFactor = bufferFactor + ) + .map(_.toSeconds) shouldBe Right(10L) + case Right(delay) => fail(s"Expected a rejection, got: $delay") + } + } + + it should "reject a frequency that is genuinely too low with an actionable message" in { + Delay.calculateOptimal(Frequency(TransformTimeUnit.Seconds, 8), nbStages = 1) match { + case Left(error) => + error should include("too small") + error should include("Minimum required frequency: 20 seconds") + error should not include "Number of stages must be positive" + case Right(delay) => fail(s"Expected a rejection, got: $delay") + } + } + + it should "keep multi-stage chains inside the latency invariant" in { + val frequency = Frequency(TransformTimeUnit.Minutes, 2) // 120s + Delay.calculateOptimal(frequency, nbStages = 3) match { + case Right(delay) => + delay.toSeconds shouldBe 20L + Delay.validate(delay, frequency, 3) shouldBe Right(()) + case Left(error) => fail(s"Expected a delay, got: $error") + } + } + + it should "agree with validate for every stage count it accepts" in { + // 600s is above the floor for all of 1..10 stages (600 / (2 * 10) = 30 >= 10), so every one of + // them MUST be accepted — asserting Right rather than tolerating a Left keeps this from passing + // against an implementation that rejects everything. + val frequency = Frequency(TransformTimeUnit.Minutes, 10) // 600s + (1 to 10).foreach { stages => + withClue(s"stages=$stages: ") { + Delay.calculateOptimal(frequency, stages) match { + case Right(delay) => + withClue(s"delay=${delay.toSeconds}s: ") { + Delay.validate(delay, frequency, stages) shouldBe Right(()) + } + case Left(error) => fail(s"Expected a delay, got: $error") + } + } + } + } +}