fix: support single-table materialized views (R1FIX.5)

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,

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,14 +126,42 @@ 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
 
 ```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,

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))

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 {

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")
+        }
+      }
+    }
+  }
+}