dongjoon-hyun commented on code in PR #58595:
URL: https://github.com/apache/spark/pull/58595#discussion_r3955396713
##########
mllib/src/main/scala/org/apache/spark/mllib/clustering/BisectingKMeansModel.scala:
##########
@@ -156,15 +158,40 @@ object BisectingKMeansModel extends
Loader[BisectingKMeansModel] {
}
}
- private def buildTree(rootId: Int, nodes: Map[Int, Data]):
ClusteringTreeNode = {
- val root = nodes(rootId)
- if (root.children.isEmpty) {
- new ClusteringTreeNode(root.index, root.size, new
VectorWithNorm(root.center, root.norm),
- root.cost, root.height, new Array[ClusteringTreeNode](0))
- } else {
- val children = root.children.map(c => buildTree(c, nodes))
- new ClusteringTreeNode(root.index, root.size, new
VectorWithNorm(root.center, root.norm),
- root.cost, root.height, children.toArray)
+ private def buildTree(rootId: Int, nodes: Map[Int, Data]):
ClusteringTreeNode =
+ buildTree(rootId, nodes, mutable.Set.empty, mutable.Map.empty)
+
+ /**
+ * `visiting` tracks the node ids currently on the recursion stack and
`built` memoizes nodes
+ * already constructed. A saved model whose child ids form a cycle (which
never happens for a
+ * valid tree) would otherwise recurse until the driver hits a
StackOverflowError; detecting the
+ * cycle turns that into a clear failure. Memoization additionally prevents
a model that
+ * references the same child id from many parents (a DAG) from expanding
exponentially. Valid
+ * trees have neither cycles nor shared children, so their construction is
unchanged.
+ */
+ private def buildTree(
+ rootId: Int,
+ nodes: Map[Int, Data],
+ visiting: mutable.Set[Int],
+ built: mutable.Map[Int, ClusteringTreeNode]): ClusteringTreeNode = {
+ built.get(rootId) match {
+ case Some(node) => node
+ case None =>
+ require(visiting.add(rootId),
+ s"Cycle detected among node ids while loading the bisecting k-means
model " +
+ s"(node id = $rootId).")
+ val root = nodes(rootId)
+ val result = if (root.children.isEmpty) {
+ new ClusteringTreeNode(root.index, root.size, new
VectorWithNorm(root.center, root.norm),
+ root.cost, root.height, new Array[ClusteringTreeNode](0))
+ } else {
+ val children = root.children.map(c => buildTree(c, nodes, visiting,
built))
+ new ClusteringTreeNode(root.index, root.size, new
VectorWithNorm(root.center, root.norm),
+ root.cost, root.height, children.toArray)
+ }
+ visiting -= rootId
+ built(rootId) = result
Review Comment:
The `built` memo silently *accepts* a DAG -- node data where several parents
reference the same child id -- and makes those parents share one
`ClusteringTreeNode` instance. But a DAG is not a valid bisecting k-means tree
either; it is exactly the same kind of inconsistent node data as a cycle. So
this change fails loudly on one form of corruption and quietly normalizes
another, which does not match the PR description ("node data is inconsistent
should fail with a clear error").
I would prefer rejecting shared children as well:
```scala
private def buildTree(
rootId: Int,
nodes: Map[Int, Data],
visiting: mutable.Set[Int]): ClusteringTreeNode = {
if (!visiting.add(rootId)) {
throw new IllegalArgumentException(
s"Cycle detected among node ids while loading the bisecting k-means
model " +
s"(node id = $rootId).")
}
...
}
```
with `visiting` never removed, so a repeated id fails whether it is a back
edge or a cross edge. That drops the memo entirely, removes the
exponential-blowup concern along with it, and makes this method symmetric with
`DecisionTreeModel.constructNode`.
If you would rather keep DAGs loadable, then please reword the comment and
the PR description so the memo is described as a performance guard rather than
part of the fail-fast behaviour.
##########
mllib/src/main/scala/org/apache/spark/mllib/tree/model/DecisionTreeModel.scala:
##########
@@ -279,31 +279,40 @@ object DecisionTreeModel extends
Loader[DecisionTreeModel] with Logging {
val dataMap: Map[Int, NodeData] = data.map(n => n.nodeId -> n).toMap
assert(dataMap.contains(1),
s"DecisionTree missing root node (id = 1).")
- constructNode(1, dataMap, mutable.Map.empty)
+ constructNode(1, dataMap, mutable.Map.empty, mutable.Set.empty)
}
/**
* Builds a node from the node data map and adds new nodes to the input
nodes map.
+ *
+ * `visiting` tracks the node ids currently on the recursion stack. A
saved model whose
+ * left/right node ids form a cycle (which never happens for a valid tree)
would otherwise
+ * recurse until the driver hits a StackOverflowError; detecting the cycle
turns that into a
+ * clear, contained failure. Valid models (including shared leaves) are
unaffected.
*/
private def constructNode(
id: Int,
dataMap: Map[Int, NodeData],
- nodes: mutable.Map[Int, Node]): Node = {
+ nodes: mutable.Map[Int, Node],
+ visiting: mutable.Set[Int]): Node = {
if (nodes.contains(id)) {
return nodes(id)
}
+ require(visiting.add(id),
Review Comment:
Minor: hiding the mutation inside the `require` predicate is easy to miss,
since `Predef.require` reads as pure argument validation. An explicit form
states the intent better:
```scala
if (!visiting.add(id)) {
throw new IllegalArgumentException(
s"Cycle detected among node ids while loading the decision tree model
(node id = $id).")
}
```
##########
mllib/src/test/scala/org/apache/spark/mllib/clustering/BisectingKMeansSuite.scala:
##########
@@ -17,14 +17,51 @@
package org.apache.spark.mllib.clustering
+import org.apache.hadoop.fs.{FileUtil, Path => HadoopPath}
+
import org.apache.spark.SparkFunSuite
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.util.MLlibTestSparkContext
import org.apache.spark.mllib.util.TestingUtils._
+import org.apache.spark.sql.functions.{array, col, lit, when}
import org.apache.spark.util.Utils
class BisectingKMeansSuite extends SparkFunSuite with MLlibTestSparkContext {
+ test("load rejects a model whose node ids form a cycle") {
Review Comment:
Same here -- please move this next to `test("BisectingKMeans model
save/load")` at the bottom of the suite rather than putting it ahead of
`test("default values")`.
##########
mllib/src/test/scala/org/apache/spark/mllib/clustering/BisectingKMeansSuite.scala:
##########
@@ -17,14 +17,51 @@
package org.apache.spark.mllib.clustering
+import org.apache.hadoop.fs.{FileUtil, Path => HadoopPath}
+
import org.apache.spark.SparkFunSuite
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.util.MLlibTestSparkContext
import org.apache.spark.mllib.util.TestingUtils._
+import org.apache.spark.sql.functions.{array, col, lit, when}
import org.apache.spark.util.Utils
class BisectingKMeansSuite extends SparkFunSuite with MLlibTestSparkContext {
+ test("load rejects a model whose node ids form a cycle") {
+ val srcDir = Utils.createTempDir()
+ val dstDir = Utils.createTempDir()
+ try {
+ val src = srcDir.toURI.toString
+ val dst = dstDir.toURI.toString
+ val data = sc.parallelize((1 until 8).map(i =>
Vectors.dense(i.toDouble)), 2)
+ val model = new BisectingKMeans().run(data)
+ model.save(sc, src)
+ val rootId = model.root.index
+
+ // Copy the metadata verbatim, and write a modified data set where the
root node lists
+ // itself as its own child (a cycle). Reading src and writing dst avoids
a self-overwrite.
+ val hadoopConf = sc.hadoopConfiguration
+ val srcMeta = new HadoopPath(s"$src/metadata")
+ val dstMeta = new HadoopPath(s"$dst/metadata")
+ FileUtil.copy(
+ srcMeta.getFileSystem(hadoopConf), srcMeta,
+ dstMeta.getFileSystem(hadoopConf), dstMeta, false, hadoopConf)
+ spark.read.parquet(s"$src/data")
+ .withColumn("children",
Review Comment:
`Loader` is `private[mllib]` and reachable from here, so please use
`Loader.metadataPath(src)` / `Loader.dataPath(src)` instead of hard-coding
`s"$src/metadata"` and `s"$src/data"` (same in `DecisionTreeSuite`). `Loader`
is the single source of truth for the on-disk layout; duplicating it as string
literals means the test would silently stop exercising anything if the layout
ever changes.
More generally, ~35 lines of parquet round-trip and Hadoop `FileUtil.copy`
plumbing per suite is quite heavy for a two-line guard. If there is a cheaper
way to reach `buildTree` / `constructNode` with crafted node data, that would
be easier to maintain.
##########
mllib/src/main/scala/org/apache/spark/mllib/tree/model/DecisionTreeModel.scala:
##########
@@ -279,31 +279,40 @@ object DecisionTreeModel extends
Loader[DecisionTreeModel] with Logging {
val dataMap: Map[Int, NodeData] = data.map(n => n.nodeId -> n).toMap
assert(dataMap.contains(1),
s"DecisionTree missing root node (id = 1).")
- constructNode(1, dataMap, mutable.Map.empty)
+ constructNode(1, dataMap, mutable.Map.empty, mutable.Set.empty)
}
/**
* Builds a node from the node data map and adds new nodes to the input
nodes map.
+ *
+ * `visiting` tracks the node ids currently on the recursion stack. A
saved model whose
+ * left/right node ids form a cycle (which never happens for a valid tree)
would otherwise
+ * recurse until the driver hits a StackOverflowError; detecting the cycle
turns that into a
+ * clear, contained failure. Valid models (including shared leaves) are
unaffected.
*/
private def constructNode(
id: Int,
dataMap: Map[Int, NodeData],
- nodes: mutable.Map[Int, Node]): Node = {
+ nodes: mutable.Map[Int, Node],
+ visiting: mutable.Set[Int]): Node = {
if (nodes.contains(id)) {
return nodes(id)
}
+ require(visiting.add(id),
+ s"Cycle detected among node ids while loading the decision tree model
(node id = $id).")
val data = dataMap(id)
val node =
if (data.isLeaf) {
Node(data.nodeId, data.predict.toPredict, data.impurity, data.isLeaf)
} else {
- val leftNode = constructNode(data.leftNodeId.get, dataMap, nodes)
- val rightNode = constructNode(data.rightNodeId.get, dataMap, nodes)
+ val leftNode = constructNode(data.leftNodeId.get, dataMap, nodes,
visiting)
+ val rightNode = constructNode(data.rightNodeId.get, dataMap, nodes,
visiting)
val stats = new InformationGainStats(data.infoGain.get,
data.impurity, leftNode.impurity,
rightNode.impurity, leftNode.predict, rightNode.predict)
new Node(data.nodeId, data.predict.toPredict, data.impurity,
data.isLeaf,
data.split.map(_.toSplit), Some(leftNode), Some(rightNode),
Some(stats))
}
+ visiting -= id
Review Comment:
Nit: this line is unreachable in effect. The node is put into `nodes` on the
next line, and every later visit to `id` short-circuits on `nodes.contains(id)`
before `visiting` is ever consulted again, so nothing can observe the removal.
Dropping it removes the need to maintain the "`visiting` == current path"
invariant at all.
##########
mllib/src/test/scala/org/apache/spark/mllib/tree/DecisionTreeSuite.scala:
##########
@@ -29,12 +31,53 @@ import org.apache.spark.mllib.tree.configuration.Strategy
import org.apache.spark.mllib.tree.impurity.{Entropy, Gini, Variance}
import org.apache.spark.mllib.tree.model._
import org.apache.spark.mllib.util.MLlibTestSparkContext
+import org.apache.spark.sql.functions.{col, lit, when}
import org.apache.spark.util.ArrayImplicits._
import org.apache.spark.util.Utils
class DecisionTreeSuite extends SparkFunSuite with MLlibTestSparkContext {
+ test("load rejects a model whose node ids form a cycle") {
Review Comment:
This suite is organized into sections with banner comments, and the new test
landed just *above* the `// Tests calling train()` banner, so it now belongs to
no section. It is a save/load test, so please move it into the `Tests of other
algorithm internals` section next to `test("model save/load")`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
