diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDesc.scala index 63fe2cacce8..6e331c4173a 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDesc.scala @@ -72,8 +72,6 @@ class TreePlotOpDesc extends PythonOperatorDescriptor { | |import plotly.graph_objects as go |import plotly.io - |import igraph - |from igraph import Graph, EdgeSeq |import pandas as pd |import ast | @@ -108,6 +106,66 @@ class TreePlotOpDesc extends PythonOperatorDescriptor { | ) | return annotations | + | def build_tree_layout(self, edges): + | # Tidy top-down tree: depth picks the row, a leaf takes the next + | # free column and a parent sits centred over its own children. + | labels = [] + | known = set() + | for parent, child in edges: + | for node in (parent, child): + | if node not in known: + | known.add(node) + | labels.append(node) + | + | children = {label: [] for label in labels} + | has_parent = set() + | seen = set() + | for parent, child in edges: + | if (parent, child) not in seen: + | seen.add((parent, child)) + | children[parent].append(child) + | has_parent.add(child) + | + | depth = {} + | column = {} + | claimed = {} + | placed = set() + | next_column = 0 + | + | def grow(root): + | nonlocal next_column + | placed.add(root) + | stack = [(root, 0, False)] + | while stack: + | node, level, folded = stack.pop() + | if folded: + | kids = claimed[node] + | if kids: + | column[node] = sum(column[kid] for kid in kids) / len(kids) + | else: + | column[node] = next_column + | next_column += 1 + | continue + | depth[node] = level + | # A node belongs to whichever parent reaches it first, so a + | # cycle or a shared child is never laid out twice. + | kids = [kid for kid in children[node] if kid not in placed] + | placed.update(kids) + | claimed[node] = kids + | stack.append((node, level, True)) + | for kid in reversed(kids): + | stack.append((kid, level + 1, False)) + | + | for label in labels: + | if label not in placed and label not in has_parent: + | grow(label) + | # Whatever is left sits in a cycle that no root can reach. + | for label in labels: + | if label not in placed: + | grow(label) + | # The y-axis is inverted here so the tree grows top-down. + | return labels, [(column[label], -depth[label]) for label in labels] + | | @overrides | def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]: | if table.empty: @@ -127,14 +185,10 @@ class TreePlotOpDesc extends PythonOperatorDescriptor { | yield {'html-content': self.render_error("No valid [parent, child] pairs found in column " + $edgeListColumn + ".")} | return | - | G = Graph.TupleList(edges, directed=True) - | labels = G.vs['name'] - | - | layout_algorithm = 'rt' | try: - | lay = G.layout(layout_algorithm) + | labels, coords = self.build_tree_layout(edges) | except Exception as e: - | yield {'html-content': self.render_error(f"Layout algorithm '{layout_algorithm}' failed: {e}")} + | yield {'html-content': self.render_error(f"Tree layout failed: {e}")} | return | | HORIZONTAL_DENSITY = 120 @@ -143,8 +197,8 @@ class TreePlotOpDesc extends PythonOperatorDescriptor { | MIN_WIDTH = 800 | MIN_HEIGHT = 600 | - | if len(lay.coords) > 1: - | x_coords, y_coords = zip(*lay.coords) + | if len(coords) > 1: + | x_coords, y_coords = zip(*coords) | x_range = max(x_coords) - min(x_coords) | y_range = max(y_coords) - min(y_coords) | plot_width = max(MIN_WIDTH, x_range * HORIZONTAL_DENSITY + PADDING) @@ -153,12 +207,13 @@ class TreePlotOpDesc extends PythonOperatorDescriptor { | plot_width = MIN_WIDTH | plot_height = MIN_HEIGHT | - | # Invert the y-axis to make the tree grow top-down. - | position = {k: (lay[k][0], -lay[k][1]) for k in range(len(labels))} + | position = {k: coords[k] for k in range(len(labels))} + | index_of = {label: k for k, label in enumerate(labels)} | | Xe = [] | Ye = [] - | for edge in G.get_edgelist(): + | for parent, child in edges: + | edge = (index_of[parent], index_of[child]) | Xe += [position[edge[0]][0], position[edge[1]][0], None] | Ye += [position[edge[0]][1], position[edge[1]][1], None] | diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDescSpec.scala index 68ea4e60879..490a5421c5a 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/treeplot/TreePlotOpDescSpec.scala @@ -19,12 +19,16 @@ package org.apache.texera.amber.operator.visualization.treeplot +import com.typesafe.config.ConfigFactory 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.Base64 +import java.util.concurrent.TimeUnit +import scala.util.Try class TreePlotOpDescSpec extends AnyFlatSpec with BeforeAndAfter with Matchers { @@ -56,6 +60,195 @@ class TreePlotOpDescSpec extends AnyFlatSpec with BeforeAndAfter with Matchers { val code = opDesc.generatePythonCode() assert(carries(code, "edge_pairs")) code should include("class ProcessTableOperator(UDFTableOperator)") - code should include("Graph.TupleList") + code should include("self.build_tree_layout(edges)") + } + + // igraph is GPL v2, Category X under the ASF 3rd party license policy, so the + // layout has to stay something the repository can actually ship. + it should "not reach for igraph" in { + opDesc.edgeListColumn = "edge_pairs" + opDesc.generatePythonCode().toLowerCase should not include "igraph" + } + + // Python executable resolution, following FilledAreaPlotOpDescSpec: + // 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; + // the generated module runs unmodified, and the layout is called directly so the + // positions it computes are what is read, rather than a rendered picture. + 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_tree_plot"} + |with open(sys.argv[1]) as f: + | exec(compile(f.read(), sys.argv[1], "exec"), ns) + |op = ns["ProcessTableOperator"]() + | + |cases = [ + | ("tree", [("a", "b"), ("a", "c"), ("b", "d"), ("b", "e")]), + | ("shared", [("a", "c"), ("b", "c")]), + | ("cycle", [("a", "b"), ("b", "c"), ("c", "a")]), + | ("selfloop", [("a", "a"), ("a", "b")]), + | ("forest", [("a", "b"), ("x", "y")]), + |] + | + |for cid, edges in cases: + | labels, coords = op.build_tree_layout(edges) + | placed = " ".join( + | "%s=%g,%g" % (label, x, y) for label, (x, y) in zip(labels, coords) + | ) + | print("CASE %s %s" % (cid, placed)) + |""".stripMargin + + /** `label -> (x, y)` per case, read out of the driver's output. One run for the + * whole suite: the driver lays every case out in the same process. + */ + private lazy val layouts: Map[String, Map[String, (Double, Double)]] = { + 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") + } + + opDesc.edgeListColumn = "edge_pairs" + val moduleFile = Files.createTempFile("tree_plot_op_", ".py") + val driverFile = Files.createTempFile("tree_plot_driver_", ".py") + try { + Files.write(moduleFile, opDesc.generatePythonCode().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("Layout 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 + } + "CASE (\\S+) (.*)".r + .findAllMatchIn(output) + .map { m => + m.group(1) -> m + .group(2) + .trim + .split(" ") + .filter(_.nonEmpty) + .map { entry => + val Array(label, xy) = entry.split("=", 2) + val Array(x, y) = xy.split(",", 2) + label -> (x.toDouble, y.toDouble) + } + .toMap + } + .toMap + } finally { + Files.deleteIfExists(moduleFile) + Files.deleteIfExists(driverFile) + } + } + + it should "lay a tree out top-down, with every parent centred over its own children" in { + val tree = layouts("tree") + // Depth picks the row and the axis is inverted, so a child sits below its parent. + tree.map { case (label, (_, y)) => label -> y } shouldBe + Map("a" -> 0.0, "b" -> -1.0, "c" -> -1.0, "d" -> -2.0, "e" -> -2.0) + // Leaves take consecutive free columns left to right, in the order they are reached. + tree("d")._1 shouldBe 0.0 + tree("e")._1 shouldBe 1.0 + tree("c")._1 shouldBe 2.0 + // A parent sits at the mean of its own children: b over d and e, a over b and c. + tree("b")._1 shouldBe (tree("d")._1 + tree("e")._1) / 2 + tree("a")._1 shouldBe (tree("b")._1 + tree("c")._1) / 2 + } + + it should "place a shared child once, under the parent that reaches it first" in { + val shared = layouts("shared") + shared.keySet shouldBe Set("a", "b", "c") + // c is claimed by a, so it hangs below a and b is left as a childless root. + shared("c") shouldBe (0.0, -1.0) + shared("a") shouldBe (0.0, 0.0) + shared("b")._2 shouldBe 0.0 + shared("b")._1 should not be shared("a")._1 + } + + it should "terminate on a cycle and still place every node once" in { + val cycle = layouts("cycle") + // No node is a root, so the layout starts from the first label it saw and the + // edge that closes the ring is dropped rather than followed a second time. + cycle shouldBe Map("a" -> (0.0, 0.0), "b" -> (0.0, -1.0), "c" -> (0.0, -2.0)) + } + + it should "survive a self loop" in { + val selfLoop = layouts("selfloop") + selfLoop.keySet shouldBe Set("a", "b") + selfLoop("a")._2 shouldBe 0.0 + selfLoop("b")._2 shouldBe -1.0 + } + + it should "lay each tree of a forest out beside the other" in { + val forest = layouts("forest") + forest.keySet shouldBe Set("a", "b", "x", "y") + // Two roots, each over its own child, in columns that do not overlap. + forest("a")._2 shouldBe 0.0 + forest("x")._2 shouldBe 0.0 + forest("a")._1 should not be forest("x")._1 + forest("b") shouldBe (forest("a")._1, -1.0) + forest("y") shouldBe (forest("x")._1, -1.0) } }