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
The following commit(s) were added to refs/heads/main by this push:
new 86865f3992 chore(workflow-operator): make FilledAreaPlot
disjoint-group tolerance a true 5% (#7149)
86865f3992 is described below
commit 86865f39926e82f5fba74574a79b05341f4e37b5
Author: Eugene Gu <[email protected]>
AuthorDate: Tue Aug 4 19:06:16 2026 -0700
chore(workflow-operator): make FilledAreaPlot disjoint-group tolerance a
true 5% (#7149)
### What changes were proposed in this PR?
**Background (#7146).** The Filled Area Plot operator draws stacked area
lines, one per Line Group, which only makes sense when the groups share
an x axis. `FilledAreaPlotOpDesc.performTableCheck()` therefore emits a
Python guard that suppresses the chart when too many groups have x-value
sets disjoint from the others; the comment above it documents the rule
as "more than 5 percents of the groups have disjoint sets of x
attributes". When the guard fires, the run still completes successfully
— the operator just yields fallback HTML instead of the chart, with no
warning anywhere.
**The defect.** The guard computes the threshold as `(len(grouped) //
100) * 5`. The integer division floors first, so the tolerance is 0 for
any chart with fewer than 100 line groups, and a single disjoint group
suppresses the whole chart — e.g. a 40-group chart with 1 disjoint group
(2.5%, well under the documented 5%) was suppressed. The two expressions
only agree when the group count is an exact multiple of 100. This has
been latent since the operator was introduced in #2086: an `X_values`
typo in the same block kept the accumulated x-value set from ever
growing, so the guard fired constantly for an unrelated reason; #6894
fixed that typo, which made the tolerance arithmetic the deciding factor
for the first time.
**The change.** Swap the operand order to `(len(grouped) * 5) // 100`,
plus four behavior-neutral cleanups to the same emitted block: the
per-group `set(...unique())` is hoisted into one local (was built twice
per group), `== None` becomes `is None`, the always-true `elif not ...`
becomes `else`, and the loop `break`s once the error is set.
Below 20 groups both expressions yield 0, which is correct (1/19 = 5.3%
exceeds 5%), so the fix only changes behavior at 20+ line groups.
### Any related issues, documentation, discussions?
Fixes #7146. Adjacent to #6728 / #6894, which fixed an `X_values` typo
in the same block but did not touch the tolerance arithmetic.
### How was this PR tested?
`performTableCheck()` previously had no test coverage. This PR adds 10
tests to `FilledAreaPlotOpDescSpec` (TDD: written red first, green after
the fix):
- 9 assertions on the generated guard, including the tolerance
expression, a single `.unique()` per iteration, `break` placement after
the error assignment, and that user-provided column names are never
emitted verbatim.
- 1 runtime test that executes the generated Python (real pandas groupby
+ plotly rendering; only the `pytexera` import seam is stubbed) across 9
boundary datasets: 19 groups/1 disjoint still suppressed (5.3% > 5%),
20/1 and 40/2 at exactly 5.0% render, 40/1 renders (the reported bug),
40/3 suppressed, plus all-disjoint, single-group, empty-table, and
missing-column cases. It cancels (not fails) when no python with
pandas+plotly is available.
Full spec: 20/20 passing. Red-check verified: reverting only the
tolerance line makes the runtime test fail on the 40/1 and exactly-5%
cases.
Checks from [CONTRIBUTING.md] all pass locally: `sbt
WorkflowOperator/scalafmtCheck` and
`WorkflowOperator/Test/scalafmtCheck` (no violations), `sbt
"WorkflowOperator/scalafixAll --check"` (clean), and the full `sbt
WorkflowOperator/test` module suite — 1967 succeeded, 0 failed, 2
pending.
### Was this PR authored or co-authored using generative AI tooling?
Co-authored by: Claude Code (Claude Fable 5)
---------
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../filledAreaPlot/FilledAreaPlotOpDesc.scala | 14 +-
.../filledAreaPlot/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 975cc4d892..84ca6082d2 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 5065315f17..8f93c3fd0f 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"
