From 0ad1a6c64c1f35813a8bf673f1856afe5bdd5c33 Mon Sep 17 00:00:00 2001 From: eugenegujing Date: Thu, 30 Jul 2026 16:50:06 -0700 Subject: [PATCH 1/2] fix(workflow-operator): make FilledAreaPlot disjoint-group tolerance a true 5% The generated guard computed `tolerance = (len(grouped) // 100) * 5`, which floors to 0 for any chart with fewer than 100 line groups, so a single disjoint line group suppressed the whole chart. Swap the operand order: `(len(grouped) * 5) // 100`. Behavior only changes at 20+ groups; below that both forms correctly give 0. Ride-along cleanups in the same emitted block: hoist the per-group `set(...unique())` into one local, use `is None`, turn the always-true `elif not ...` into `else`, and `break` once the verdict is final. Adds the first tests covering performTableCheck(): 9 assertions on the generated guard plus a runtime boundary test that executes the generated Python (cancels gracefully when pandas/plotly are unavailable). Fixes #7146 --- .../filledAreaPlot/FilledAreaPlotOpDesc.scala | 14 +- .../FilledAreaPlotOpDescSpec.scala | 221 ++++++++++++++++++ 2 files changed, 229 insertions(+), 6 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDesc.scala index 975cc4d8929..84ca6082d24 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDesc.scala +++ b/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 |""" } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDescSpec.scala index 5065315f175..8f93c3fd0f0 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/filledAreaPlot/FilledAreaPlotOpDescSpec.scala @@ -19,6 +19,7 @@ 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 @@ -26,6 +27,11 @@ 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" From e9ac8bad6fa8e1fa3e0c6090a808d796ff113ad7 Mon Sep 17 00:00:00 2001 From: eugenegujing Date: Thu, 30 Jul 2026 17:22:41 -0700 Subject: [PATCH 2/2] ci: retrigger checks Empty commit to rerun the amber-integration job: ReconfigurationIntegrationSpec failed once on macOS only (known flaky pause/resume race, unrelated to this change; the same tests passed on ubuntu). Co-Authored-By: Claude Fable 5