(texera) 01/02: test(workflow-operator): cover the SQL source's batch boundaries and dispatch (#7871)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git

commit fb1bb987603169b9879ad12a9578fc2a226c3945
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Sun Aug 23 05:54:55 2026 +0000

    test(workflow-operator): cover the SQL source's batch boundaries and dispatch (#7871)
    
    ### What changes were proposed in this PR?
    
    `SQLSourceOpExecSpec` goes from 30 tests to 40, covering the
    batch-boundary arithmetic and type dispatch in `SQLSourceOpExec`.
    
    Measured with the `FileScanSourceOpExecSpec` suite-name exclusion
    applied identically to the before and after run — that suite aborts on
    this machine for pre-existing reasons, and because sbt-jacoco runs
    unforked a failing test task skips `saveRuntimeData` and emits an
    all-zero report rather than a partial one.
    
    | Metric | Before | After |
    |---|---|---|
    | Codecov (fully-covered lines) | 139/163 = 85.3% | **147/163 = 90.2%**
    |
    | Missed lines | 4 (230, 231, 364, 519) | **0** |
    | JaCoCo line-hit | 159/163 = 97.6% | **163/163 = 100%** |
    | Branch arms | 127 covered / 43 missed | **137 covered / 33 missed** |
    
    Tests: spec 30 → 40; the `source.sql.*` package 176 → 186 across 11
    suites; the module 2307 → 2317.
    
    ### The +8 is not eight lines of behaviour, and the first draft implied
    it was
    
    Of the eight newly fully-covered lines, **five are production-reachable
    behaviour** — the INTEGER and DOUBLE type dispatch, the progressive arm
    of the offset guard, the non-auto LONG maximum, and the non-auto
    unsupported-type throw. The other **three are defensive arms no workflow
    can reach**: `hasNextQuery`'s unsupported-type throw is unreachable
    because a successful `open()` admits only INTEGER/LONG/TIMESTAMP/DOUBLE,
    and two more are similar guards.
    
    A reviewer put the split at +6 reachable / +2 defensive. That was closer
    than the original claim but still not right; the measured split is 5 and
    3.
    
    ### Verification
    
    22 mutations, **20 killed, 2 survivors.** Every one of the nine mutants
    the two reviewers reported as surviving was first *reproduced surviving*
    on the pre-review spec, and eight of the nine now die.
    
    **The two survivors are deliberate, and killing them would make the
    suite worse.** `isLastBatch = nextLowerBound >= upperBound` can be
    flipped to `>` on both the LONG arm (line 282) and its DOUBLE twin (line
    285) and survive all 186 package tests. This was verified by running it,
    not inferred. The only input that distinguishes the two makes production
    emit a **duplicate final window** — so a test that killed these mutants
    would cement that defect. Reported instead.
    
    ### Corrections to my own first draft
    
    - "10 mutations, all 10 killed, zero survivors" was true of that table,
    but the table was too narrow to support the bundle's claims — nine
    further mutants on the very lines the bundle claimed to newly cover were
    untested, and most of them survived.
    - The +8/+10 figures reproduce exactly, but were stated without the
    reachability qualification above.
    - The conclusion drawn from line 502 was wrong even though its coverage
    half was right.
    - Module baselines were off by two: 208 missed lines (96.59%) and 1215
    missed branches (51.96%), not 206/1214.
    - Line counts and test counts in the first draft were stale.
    - One assertion — `(statement.setLong _).expects(*, *).never()` —
    carried no information beyond the strict mock itself and is gone,
    replaced by a positive `expects(1, 2L)` whose strictness does the work.
    - A CI-wiring claim from the first draft is not restated here, because
    it was not re-verified in this pass.
    
    ### Deliberately not included, with evidence
    
    - **Four `case A | B | C | _ =>` arms** (230, 286, 320, 357) keep a dead
    `ifeq` side: scalac compiles the `_` as a constant before the final
    branch.
    - **Ten lines** are sealed-`Option` `None$.equals` checks falling
    through to `new scala/MatchError`, plus an `$outer` null check and a
    non-local-return rethrow.
    - **Line 502's `batchByAttribute.isDefined` false arm is dead**:
    `initBatchColumnBoundaries` is private with one caller, `open()`, which
    always assigns `Option(schema.getAttribute(...))`, and
    `Schema.getAttribute` throws rather than returning null. Widening the
    private would be a production edit — refused.
    - **Line 472's `if (limit > 0)` false arm is unreachable**:
    `generateSqlQuery` returns `None` whenever the limit is `Some(n <= 0)`,
    so 472 is only reached when the limit is positive. Getting there needs
    the protected `generateSqlQuery` overridden, i.e. replacing the code
    under test.
    
    No production file is touched — sha256 unchanged from the pre-mutation
    snapshot, and no stray `test_large_binary.txt` was left behind, since
    the FileScan suite never ran.
    
    ### Any related issues, documentation, discussions?
    
    Closes #7868
    
    ### How was this PR tested?
    
    ```
    sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.source.sql.SQLSourceOpExecSpec"
    ```
    
    ```
    [info] Total number of tests run: 40
    [info] Tests: succeeded 40, failed 0, canceled 0, ignored 0, pending 0
    ```
    
    The whole `source.sql.*` package is green at 186 tests across 11 suites,
    0 aborted. `Test/scalafmtCheck` passes.
    
    ### Was this PR authored or co-authored using generative AI tooling?
    
    Generated-by: Claude Code (Opus 5)
    
    ---------
    
    Signed-off-by: Xinyuan Lin <xi...@uci.edu>
    Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
 .../operator/source/sql/SQLSourceOpExecSpec.scala  | 192 ++++++++++++++++++++-
 1 file changed, 189 insertions(+), 3 deletions(-)

diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala
index 7d549b2c18..f947ba284d 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala
@@ -47,6 +47,15 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
     List(new Attribute("id", AttributeType.INTEGER), new Attribute("name", AttributeType.STRING))
   )
 
+  /** `rowSchema` plus the LONG column the progressive row tests batch by. */
+  private val progressiveRowSchema = Schema(
+    List(
+      new Attribute("id", AttributeType.INTEGER),
+      new Attribute("name", AttributeType.STRING),
+      new Attribute("big", AttributeType.LONG)
+    )
+  )
+
   /**
     * Minimal concrete executor: the DB-touching hooks are stubbed out so the
     * query-building surface can be exercised on its own.
@@ -86,6 +95,7 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
     def terminator: String = render(b => terminateSQL(b))
     def slidingWindow: String = render(b => addBatchSlidingWindow(b))
     def batchValue(value: Number): String = batchAttributeToString(value)
+    def boundary(side: String): Number = fetchBatchByBoundary(side)
   }
 
   private def descJson(configure: PostgreSQLSourceOpDesc => Unit = _ => ()): String = {
@@ -128,10 +138,17 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
     exec
   }
 
-  /** A connection answering the `SELECT MIN/MAX(col) FROM tbl;` boundary probes. */
-  private def boundaryConn(column: String, stub: (ResultSet, String) => Unit): Connection = {
+  /**
+    * A connection answering the `SELECT MIN/MAX(col) FROM tbl;` boundary probes.
+    * `sides` narrows it to the probes the executor is expected to actually issue.
+    */
+  private def boundaryConn(
+      column: String,
+      stub: (ResultSet, String) => Unit,
+      sides: Seq[String] = Seq("MIN", "MAX")
+  ): Connection = {
     val conn = mock[Connection]
-    Seq("MIN", "MAX").foreach { side =>
+    sides.foreach { side =>
       val statement = mock[PreparedStatement]
       val resultSet = mock[ResultSet]
       (conn
@@ -202,7 +219,47 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
     exec.sqlQuery shouldBe Some(BASE + ";")
   }
 
+  it should "skip the sliding window when progressive mode names no batch column" in {
+    // A progressive descriptor with no batchByColumn is degenerate but generateSqlQuery is
+    // defensive about it. open() is deliberately not called here: it would blow up on
+    // `desc.batchByColumn.get`, which is the current (unhelpful) behaviour, so this test
+    // pins only the query builder. That also means this arm of the progressive guard
+    // (progressive with no batch column) is defensive: no production caller reaches it,
+    // generateSqlQuery in this state, since open() throws first.
+    // min and max are populated on purpose: with all three of batchByColumn/min/max empty,
+    // re-pointing the guard at `desc.min` instead would be indistinguishable here.
+    val exec = new TestSQLSourceOpExec(descJson { desc =>
+      desc.progressive = Option(true)
+      desc.batchByColumn = None
+      desc.min = Option("0")
+      desc.max = Option("100")
+      desc.interval = 30L
+    })
+    exec.sqlQuery shouldBe Some(BASE + ";")
+  }
+
+  it should "ignore a leftover batch column when progressive mode is off" in {
+    // `progressive` only toggles batchByColumn/min/max/interval *hidden* in the UI
+    // (SQLSourceOpDesc's toggleHidden), it does not clear them, so a non-progressive
+    // descriptor can still carry a batch column. Neither the sliding window nor the
+    // fixed OFFSET may key off that column instead of the progressive flag.
+    val exec = new TestSQLSourceOpExec(descJson { desc =>
+      desc.batchByColumn = Option("big")
+      desc.min = Option("0")
+      desc.max = Option("100")
+      desc.interval = 30L
+    })
+    exec.curOffset = Some(5L)
+    exec.sqlQuery shouldBe Some(BASE + " OFFSET ?;")
+  }
+
   it should "walk the LONG batch column window by window until the upper bound" in {
+    // Deliberately NOT pinned: the `>=` that decides the last window (SQLSourceOpExec's
+    // `isLastBatch`, and its DOUBLE twin) is only observable when the upper bound sits
+    // exactly one interval above the current lower bound -- and on that input production
+    // emits a redundant `>= upper AND <= upper` window that re-scans the boundary row.
+    // Asserting the current behaviour there would cement that duplication, so it is
+    // reported as a defect rather than encoded here.
     val exec = openedProgressive("big", "0", "100", 30L)
 
     exec.nextQueryAvailable shouldBe true
@@ -279,6 +336,36 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
     exec.nextQueryAvailable shouldBe false
   }
 
+  it should "offer a next query while an INTEGER batch column is below its upper bound" in {
+    val conn = boundaryConn(
+      "id",
+      (rs, side) => (rs.getInt(_: Int)).expects(1).returning(if (side == "MIN") 0 else 10)
+    )
+    val exec = openedProgressive("id", "auto", "auto", 4L, conn)
+
+    exec.nextQueryAvailable shouldBe true
+    exec.slidingWindow shouldBe " AND id >= 0 AND id < 4"
+
+    // A single-value range (min == max) must still yield exactly one window. This is the
+    // one input on which the `<=` in hasNextQuery is load-bearing, and unlike the
+    // exact-multiple case it does not exercise the duplicate-last-window defect.
+    val singleValueConn =
+      boundaryConn("id", (rs, _) => (rs.getInt(_: Int)).expects(1).returning(7))
+    val singleValue = openedProgressive("id", "auto", "auto", 4L, singleValueConn)
+    singleValue.nextQueryAvailable shouldBe true
+    singleValue.slidingWindow shouldBe " AND id >= 7 AND id <= 7"
+    singleValue.nextQueryAvailable shouldBe false
+  }
+
+  it should "refuse to decide the next query for an unsupported batch column type" in {
+    val exec = new TestSQLSourceOpExec(progressiveJson("name", Option("a"), Option("z"), 4L))
+    // DEFENSIVE-ONLY arm: no workflow can reach hasNextQuery with a STRING batch column, but
+    // this pins the type-dispatch error message without depending on open()'s failure path.
+    exec.batchByAttribute = Some(new Attribute("name", AttributeType.STRING))
+    intercept[IllegalArgumentException](exec.nextQueryAvailable).getMessage shouldBe
+      "Unexpected type: string"
+  }
+
   "SQLSourceOpExec.open" should "validate the table against the loaded table names" in {
     val exec = new TestSQLSourceOpExec(descJson(), knownTables = Seq("other"))
     val thrown = intercept[RuntimeException](exec.open())
@@ -286,6 +373,20 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
     exec.loadTableNamesCalls shouldBe 1
   }
 
+  it should "validate the table before probing the batch column boundaries" in {
+    // fetchBatchByBoundary concatenates desc.table straight into the probe SQL, so the
+    // table-name check -- which this file documents as its SQL-injection defence -- has to
+    // run before any boundary probe. The strict mock with no expectations is the oracle:
+    // any probe at all is an unexpected call.
+    val conn = mock[Connection]
+    val exec = new TestSQLSourceOpExec(
+      progressiveJson("big", Option("auto"), Option("auto"), 30L),
+      conn,
+      knownTables = Seq("other")
+    )
+    intercept[RuntimeException](exec.open()).getMessage shouldBe "Can't find the given table `tbl`."
+  }
+
   it should "leave batchByAttribute unset when progressive mode is disabled" in {
     val exec = new TestSQLSourceOpExec(descJson())
     exec.open()
@@ -299,6 +400,34 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
     thrown.getMessage should startWith("Missing required progressive configuration")
   }
 
+  it should "reject progressive mode with a lower but no upper boundary" in {
+    val exec = new TestSQLSourceOpExec(progressiveJson("big", Option("0"), None, 30L))
+    val thrown = intercept[IllegalArgumentException](exec.open())
+    thrown.getMessage should startWith("Missing required progressive configuration")
+  }
+
+  it should "reject progressive mode with an upper but no lower boundary" in {
+    // the mirror of the case above: without this one, dropping the min check from the
+    // configuration guard would go unnoticed and desc.min.get would raise a bare
+    // NoSuchElementException instead of this message
+    val exec = new TestSQLSourceOpExec(progressiveJson("big", None, Option("100"), 30L))
+    val thrown = intercept[IllegalArgumentException](exec.open())
+    thrown.getMessage should startWith("Missing required progressive configuration")
+  }
+
+  it should "reject a non-auto upper bound on a column that is neither LONG nor TIMESTAMP" in {
+    // the MIN side resolves through the auto probe, so the MAX side is the one that
+    // reaches the type match and rejects the DOUBLE column
+    val conn = boundaryConn(
+      "score",
+      (rs, _) => (rs.getDouble(_: Int)).expects(1).returning(1.5),
+      sides = Seq("MIN")
+    )
+    val exec =
+      new TestSQLSourceOpExec(progressiveJson("score", Option("auto"), Option("5"), 4L), conn)
+    intercept[IllegalArgumentException](exec.open()).getMessage shouldBe "Unsupported type double"
+  }
+
   it should "fetch auto boundaries for every supported batch column type" in {
     val intConn = boundaryConn(
       "id",
@@ -338,6 +467,15 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
     exec.slidingWindow shouldBe " AND score >= 1.5 AND score < 5.5"
     exec.slidingWindow shouldBe " AND score >= 5.5 AND score <= 9.0"
     exec.nextQueryAvailable shouldBe false
+
+    // the DOUBLE mirror of the INTEGER single-value case: min == max must still yield one
+    // window, which is the only input that makes this arm's `<=` load-bearing
+    val singleValueConn =
+      boundaryConn("score", (rs, _) => (rs.getDouble(_: Int)).expects(1).returning(1.5))
+    val singleValue = openedProgressive("score", "auto", "auto", 4L, singleValueConn)
+    singleValue.nextQueryAvailable shouldBe true
+    singleValue.slidingWindow shouldBe " AND score >= 1.5 AND score <= 1.5"
+    singleValue.nextQueryAvailable shouldBe false
   }
 
   it should "reject an auto boundary probe on an unsupported column type" in {
@@ -360,6 +498,15 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
     intercept[IllegalStateException](exec.open()).getMessage shouldBe "Unexpected value: string"
   }
 
+  it should "report a zero boundary when no batch column is configured" in {
+    // DEFENSIVE-ONLY arm, not a reachable execution path: fetchBatchByBoundary is called from
+    // exactly two production sites, both inside initBatchColumnBoundaries behind
+    // `batchByAttribute.isDefined`, and AsterixDBSourceOpExec overrides the method outright.
+    // What this pins is the protected contract subclasses inherit, nothing a workflow can run.
+    // no connection is needed: without a batchByAttribute the probe never touches JDBC
+    new TestSQLSourceOpExec(descJson()).boundary("MIN").intValue shouldBe 0
+  }
+
   "SQLSourceOpExec.produceTuple" should "stream every row of the single non-progressive query" in {
     val resultSet = mock[ResultSet]
     val statement = mock[PreparedStatement]
@@ -435,6 +582,45 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
     exec.curLimit shouldBe Some(1L)
   }
 
+  it should "bind no fixed offset in progressive mode" in {
+    val resultSet = mock[ResultSet]
+    val statement = mock[PreparedStatement]
+    val conn = mock[Connection]
+
+    // the limit is present so that progressive-ness and "has a limit" are distinguishable
+    // here, and so that the clause order (sliding window ahead of LIMIT) is pinned
+    (conn
+      .prepareStatement(_: String))
+      .expects(BASE + " AND big >= 0 AND big < 30 LIMIT ?;")
+      .returning(statement)
+    // only the limit is bound, at index 1: progressive mode carries its offset in the
+    // sliding window, and the strict mock rejects any further setLong, i.e. any offset
+    (statement.setLong _).expects(1, 2L)
+    (statement.executeQuery: () => ResultSet).expects().returning(resultSet)
+    inSequence {
+      (resultSet.next _).expects().returning(true) // consumed by the manual offset skip
+      (resultSet.next _).expects().returning(true)
+    }
+    (resultSet.getObject(_: String)).expects("id").returning(Int.box(4))
+    (resultSet.getObject(_: String)).expects("name").returning("p")
+    (resultSet.getObject(_: String)).expects("big").returning(Long.box(7L))
+
+    val exec = new TestSQLSourceOpExec(
+      progressiveJson("big", Option("0"), Option("100"), 30L),
+      conn,
+      execSchema = progressiveRowSchema
+    )
+    exec.curOffset = Some(1L)
+    exec.curLimit = Some(2L)
+    exec.open()
+
+    val first = exec.produceTuple().next().asInstanceOf[Tuple]
+    first.getField[Any]("id") shouldBe 4
+    first.getField[Any]("big") shouldBe 7L
+    exec.curOffset shouldBe Some(0L)
+    exec.curLimit shouldBe Some(1L)
+  }
+
   it should "yield nothing when the remaining limit already forbids a query" in {
     val conn = mock[Connection]
     val exec = new TestSQLSourceOpExec(descJson(), conn, execSchema = rowSchema)