feat(treeplot): lay out the tree without igraph

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]
        |

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)
   }
 }