fix(workflow-operator): make FilledAreaPlot disjoint-group tolerance a true 5%

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDesc.scala

@@ -114,18 +114,20 @@ class FilledAreaPlotOpDesc extends PythonOperatorDescriptor {
        |            grouped = table.groupby($lineGroup)
        |            x_values = None
        |
-       |            tolerance = (len(grouped) // 100) * 5
+       |            tolerance = (len(grouped) * 5) // 100
        |            count = 0
        |
        |            for _, group in grouped:
-       |                if x_values == None:
-       |                    x_values = set(group[$x].unique())
-       |                elif set(group[$x].unique()).intersection(x_values):
-       |                    x_values = x_values.union(set(group[$x].unique()))
-       |                elif not set(group[$x].unique()).intersection(x_values):
+       |                group_x_values = set(group[$x].unique())
+       |                if x_values is None:
+       |                    x_values = group_x_values
+       |                elif group_x_values.intersection(x_values):
+       |                    x_values = x_values.union(group_x_values)
+       |                else:
        |                    count += 1
        |                    if count > tolerance:
        |                        error = "X attributes not shared across groups"
+       |                        break
        |"""
   }
 

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDescSpec.scala

@@ -19,13 +19,19 @@
 
 package org.apache.texera.amber.operator.visualization.filledAreaPlot
 
+import com.typesafe.config.ConfigFactory
 import org.apache.texera.amber.core.tuple.{AttributeType, Schema}
 import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.scalatest.BeforeAndAfter
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+import scala.util.Try
+
 class FilledAreaPlotOpDescSpec extends AnyFlatSpec with BeforeAndAfter with Matchers {
 
   var opDesc: FilledAreaPlotOpDesc = _
@@ -131,6 +137,221 @@ class FilledAreaPlotOpDescSpec extends AnyFlatSpec with BeforeAndAfter with Matc
     plain should include("pattern_shape=")
   }
 
+  private def generatedGuardCode(): String = {
+    opDesc.x = "x"
+    opDesc.y = "y"
+    opDesc.lineGroup = "g"
+    opDesc.generatePythonCode()
+  }
+
+  "FilledAreaPlotOpDesc.generatePythonCode" should
+    "compute the disjoint-group tolerance as five percent of the group count" in {
+    val code = generatedGuardCode()
+    code should include("(len(grouped) * 5) // 100")
+    code should not include "(len(grouped) // 100) * 5"
+  }
+
+  it should "collect each group's unique x values exactly once per loop iteration" in {
+    val code = generatedGuardCode()
+    "\\.unique\\(\\)".r.findAllIn(code).size shouldBe 1
+  }
+
+  it should "compare x_values against None with an identity check" in {
+    val code = generatedGuardCode()
+    code should include("x_values is None")
+    code should not include "x_values == None"
+  }
+
+  it should "treat the disjoint-group branch as a plain else" in {
+    val code = generatedGuardCode()
+    code should not include "elif not set("
+    code should include("else:")
+  }
+
+  it should "break out of the group loop once the error is set" in {
+    val code = generatedGuardCode()
+    code should include("break")
+  }
+
+  it should "keep the shared-x guard with its tolerance and error message" in {
+    val code = generatedGuardCode()
+    code should include("tolerance")
+    code should include("X attributes not shared across groups")
+  }
+
+  it should "never emit user-provided column names verbatim" in {
+    opDesc.x = "distinctive_col_xyz"
+    opDesc.y = "distinctive_col_abc"
+    opDesc.lineGroup = "distinctive_grp_qrs"
+    val code = opDesc.generatePythonCode()
+    code should include("decode_python_template")
+    code should not include "distinctive_col_xyz"
+    code should not include "distinctive_col_abc"
+    code should not include "distinctive_grp_qrs"
+  }
+
+  it should "place the break after the disjoint-group error assignment" in {
+    val code = generatedGuardCode()
+    val errorIndex = code.indexOf("error = \"X attributes not shared across groups\"")
+    val breakIndex = code.indexOf("break")
+    errorIndex should be > -1
+    breakIndex should be > errorIndex
+  }
+
+  it should "emit the missing-attributes branch with its fallback text" in {
+    val code = generatedGuardCode()
+    code should include("error = \"missing attributes\"")
+    code should include("X or Y attribute does not exist")
+  }
+
+  // Python executable resolution, following PythonCodeRawInvalidTextSpec:
+  // udf.conf python.path (UDF_PYTHON_PATH), then python3 / python / py.
+  private def resolvePythonExecutable(): Option[String] = {
+    def fromConfig: Option[String] = {
+      val configOpt =
+        Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption
+          .orElse(Try(ConfigFactory.load()).toOption)
+      configOpt
+        .flatMap(c => Try(c.getConfig("python").getString("path")).toOption)
+        .map(_.trim)
+        .filter(_.nonEmpty)
+    }
+
+    def isRunnable(exe: String): Boolean = {
+      val pTry = Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start())
+      pTry.toOption.exists { p =>
+        val finished = p.waitFor(5, TimeUnit.SECONDS)
+        if (!finished) { p.destroyForcibly(); false }
+        else p.exitValue() == 0
+      }
+    }
+
+    (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(isRunnable)
+  }
+
+  private def canImportPandasAndPlotly(python: String): Boolean = {
+    val pTry = Try(
+      new ProcessBuilder(python, "-c", "import pandas, plotly").redirectErrorStream(true).start()
+    )
+    pTry.toOption.exists { p =>
+      val finished = p.waitFor(60, TimeUnit.SECONDS)
+      if (!finished) { p.destroyForcibly(); false }
+      else p.exitValue() == 0
+    }
+  }
+
+  // Driver executed by the runtime test below. It stubs only the pytexera import seam
+  // (base class, decorator, and type aliases); the generated module itself runs unmodified,
+  // including the real pandas groupby guard and plotly rendering.
+  private val runtimeDriverScript: String =
+    """import base64
+      |import sys
+      |import types
+      |from typing import Iterator, Optional
+      |
+      |import pandas as pd
+      |
+      |class UDFTableOperator:
+      |    def decode_python_template(self, data):
+      |        return base64.b64decode(data).decode("utf-8")
+      |
+      |stub = types.ModuleType("pytexera")
+      |stub.UDFTableOperator = UDFTableOperator
+      |stub.overrides = lambda fn: fn
+      |stub.Table = pd.DataFrame
+      |stub.TableLike = object
+      |stub.Iterator = Iterator
+      |stub.Optional = Optional
+      |sys.modules["pytexera"] = stub
+      |
+      |ns = {"__name__": "generated_filled_area_plot"}
+      |with open(sys.argv[1]) as f:
+      |    exec(compile(f.read(), sys.argv[1], "exec"), ns)
+      |op = ns["ProcessTableOperator"]()
+      |
+      |def make_df(n_groups, disjoint):
+      |    rows = []
+      |    for i in range(n_groups):
+      |        xs = range(101, 111) if i in disjoint else range(1, 11)
+      |        rows += [{"g": "g%03d" % i, "x": x, "y": float(x) * (i + 1)} for x in xs]
+      |    return pd.DataFrame(rows, columns=["g", "x", "y"])
+      |
+      |all_disjoint = pd.DataFrame(
+      |    [{"g": "g%03d" % i, "x": x, "y": float(x)}
+      |     for i in range(5) for x in range(i * 100 + 1, i * 100 + 11)]
+      |)
+      |
+      |cases = [
+      |    ("b19_1", make_df(19, {18})),
+      |    ("b20_1", make_df(20, {19})),
+      |    ("b40_1", make_df(40, {39})),
+      |    ("b40_2", make_df(40, {38, 39})),
+      |    ("b40_3", make_df(40, {37, 38, 39})),
+      |    ("alldis", all_disjoint),
+      |    ("single", make_df(1, set())),
+      |    ("empty", pd.DataFrame({"x": [], "y": [], "g": []})),
+      |    ("miscol", pd.DataFrame({"wrong": [1], "y": [1.0], "g": ["g0"]})),
+      |]
+      |
+      |for cid, df in cases:
+      |    html = list(op.process_table(df, 0))[0]["html-content"]
+      |    if "not shared across all line groups" in html:
+      |        verdict = "FALLBACK"
+      |    elif "does not exist" in html:
+      |        verdict = "MISSING"
+      |    else:
+      |        verdict = "CHART"
+      |    print("CASE %s %s" % (cid, verdict))
+      |""".stripMargin
+
+  it should "enforce the five-percent tolerance at runtime boundaries" in {
+    val python = resolvePythonExecutable().getOrElse(
+      cancel("No runnable python executable (udf.conf python.path, python3, python, py)")
+    )
+    if (!canImportPandasAndPlotly(python)) {
+      cancel(s"'$python' cannot import pandas and plotly; skipping runtime verification")
+    }
+
+    val moduleFile = Files.createTempFile("filled_area_plot_op_", ".py")
+    val driverFile = Files.createTempFile("filled_area_plot_driver_", ".py")
+    try {
+      Files.write(moduleFile, generatedGuardCode().getBytes(StandardCharsets.UTF_8))
+      Files.write(driverFile, runtimeDriverScript.getBytes(StandardCharsets.UTF_8))
+
+      val process = new ProcessBuilder(python, driverFile.toString, moduleFile.toString)
+        .redirectErrorStream(true)
+        .start()
+      val finished = process.waitFor(120, TimeUnit.SECONDS)
+      if (!finished) {
+        process.destroyForcibly()
+        fail("Runtime verification driver timed out after 120s")
+      }
+      val output = new String(process.getInputStream.readAllBytes(), StandardCharsets.UTF_8)
+      withClue(s"Driver output:\n$output\n") {
+        process.exitValue() shouldBe 0
+        val verdicts = "CASE (\\S+) (\\S+)".r
+          .findAllMatchIn(output)
+          .map(m => m.group(1) -> m.group(2))
+          .toMap
+        verdicts shouldBe Map(
+          "b19_1" -> "FALLBACK", // 1/19 disjoint = 5.3% > 5%
+          "b20_1" -> "CHART", // exactly 5.0%, tolerance is strict >
+          "b40_1" -> "CHART", // 2.5%: the original floored-tolerance bug
+          "b40_2" -> "CHART", // exactly 5.0%
+          "b40_3" -> "FALLBACK", // 7.5%
+          "alldis" -> "FALLBACK", // every group disjoint
+          "single" -> "CHART", // one group, nothing to compare
+          "empty" -> "CHART", // no rows, guard loop never runs
+          "miscol" -> "MISSING" // x column absent
+        )
+      }
+    } finally {
+      Try(Files.deleteIfExists(moduleFile))
+      Try(Files.deleteIfExists(driverFile))
+      ()
+    }
+  }
+
   "FilledAreaPlotOpDesc" should "round-trip its config fields through the polymorphic base" in {
     opDesc.x = "area_x"
     opDesc.y = "area_y"