RangeSetND: speed up multivariate folding

doc/sphinx/tools/nodeset.rst

@@ -565,6 +565,38 @@ contiguous node set in an expanded manner to separate lines::
     node11 node12 node13 node14 node15 node16 node17 node18 node19
 
 
+How nD folding works
+""""""""""""""""""""
+
+Internally, a multidimensional node set is stored as a list of *range
+vectors*, one range set per axis: ``x[1-2]c[1-4]`` is the vector
+``(1-2; 1-4)`` and represents the Cartesian product of its axes, that is
+2 x 4 = 8 nodes. When node sets are combined, ClusterShell *folds* the
+result in two steps: it first rewrites it as vectors that do not overlap,
+then repeatedly merges any two vectors that are identical on all axes but
+one — the only kind of merge that is always exact (for example,
+``x1c[1-4]`` and ``x2c[1-4]`` merge into ``x[1-2]c[1-4]``). Vectors are
+finally sorted, larger first. The folded output is canonical: it depends
+only on the resulting set of nodes, not on the order in which node sets
+were combined or expressed.
+
+Folding aims for compact output but does not guarantee the smallest
+possible number of vectors, as computing an optimal decomposition is a
+hard combinatorial problem. The exact decomposition of an equivalent node
+set may also vary between ClusterShell releases, so scripts should compare
+node sets with set operations rather than by their string representations.
+For example, to check that two node sets are equal, use ``-X`` (symmetric
+difference) to count the nodes that are in one set but not in the other::
+
+    $ nodeset -c "x1c[1-4],x2c[1-4]" -X "x[1-2]c[1-4]"
+    0
+
+A count of 0 means the two node sets contain exactly the same nodes. With
+``-f`` instead of ``-c``, the command prints the differing nodes, if any::
+
+    $ nodeset -f "x1c[1-4],x2c[1-4]" -X "x[1-2]c[1-5]"
+    x[1-2]c5
+
 Choosing fold axis (nD)
 """""""""""""""""""""""
 

lib/ClusterShell/RangeSet.py

@@ -1265,8 +1265,13 @@ def rgveckeyfunc(rgvec):
             #   (2) larger dim first  (#elements)
             #   (3) lower first index first
             #   (4) lower last index first
-            return (-reduce(mul, [len(rg) for rg in rgvec]), \
-                    tuple((-len(rg), rg[0], rg[-1]) for rg in rgvec))
+            size = 1
+            subkeys = []
+            for rg in rgvec:
+                srtvec = rg._sorted()  # call _sorted() only once per axis
+                size *= len(srtvec)
+                subkeys.append((-len(srtvec), srtvec[0], srtvec[-1]))
+            return (-size, tuple(subkeys))
         self._veclist.sort(key=rgveckeyfunc)
 
     @precond_fold()
@@ -1319,69 +1324,72 @@ def _fold_multivariate(self):
         self._dirty = False
 
     def _fold_multivariate_expand(self):
-        """Multivariate nD folding: expand [phase 1]"""
-        self._veclist = [[RangeSet.fromone(i, autostep=self.autostep)
-                          for i in tvec]
-                         for tvec in set(self._iter())]
+        """Multivariate nD folding: expand [phase 1]
+
+        Group unique elements by leading coordinates, building maximal
+        disjoint vectors along the last axis."""
+        rows = {}
+        for rgvec in self._veclist:
+            last = rgvec[-1]
+            if not last:
+                continue
+            for prefix in product(*rgvec[:-1]):
+                row = rows.get(prefix)
+                if row is None:
+                    rows[prefix] = set(last)
+                else:
+                    row.update(last)
+
+        veclist = []
+        for prefix, elems in rows.items():
+            rgvec = []
+            for coord in prefix:
+                rg = RangeSet()
+                rg._autostep = self._autostep
+                set.add(rg, coord)  # fresh set
+                rgvec.append(rg)
+            rg = RangeSet()
+            rg._autostep = self._autostep
+            set.update(rg, elems)  # fresh set
+            rgvec.append(rg)
+            veclist.append(rgvec)
+        self._veclist = veclist
 
     def _fold_multivariate_merge(self):
-        """Multivariate nD folding: merge [phase 2]"""
-        full = False  # try easy O(n) passes first
-        chg = True    # new pass (eg. after change on veclist)
+        """Multivariate nD folding: merge [phase 2]
+
+        Merge vectors that differ by only one axis: group them by their
+        other axes (hashable frozenset keys), one axis at a time, until
+        no more merges are found. Expanded vectors are always disjoint,
+        so merging is a simple disjoint union on the differing axis."""
+        veclist = self._veclist
+        dim = self.dim()
+        # skip last axis on first pass, it was already merged by expand
+        startpos = dim - 2
+        chg = len(veclist) > 1
         while chg:
             chg = False
-            self._sort()  # sort veclist before new pass
-            index1, index2 = 0, 1
-            while (index1 + 1) < len(self._veclist):
-                # use 2 references on iterator to compare items by couples
-                item1 = self._veclist[index1]
-                index2 = index1 + 1
-                index1 += 1
-                while index2 < len(self._veclist):
-                    item2 = self._veclist[index2]
-                    index2 += 1
-                    new_item = [None] * len(item1)
-                    nb_diff = 0
-                    # compare 2 rangeset vector, item by item, the idea being
-                    # to merge vectors if they differ only by one item
-                    for pos, (rg1, rg2) in enumerate(zip(item1, item2)):
-                        if rg1 == rg2:
-                            new_item[pos] = rg1
-                        elif not rg1 & rg2: # merge on disjoint ranges
-                            nb_diff += 1
-                            if nb_diff > 1:
-                                break
-                            new_item[pos] = rg1 | rg2
-                        # if fully contained, keep the largest one
-                        elif (rg1 > rg2 or rg1 < rg2): # and nb_diff == 0:
-                            nb_diff += 1
-                            if nb_diff > 1:
-                                break
-                            new_item[pos] = max(rg1, rg2)
-                        # otherwise, compute rangeset intersection and
-                        # keep the two disjoint part to be handled
-                        # later...
-                        else:
-                            # intersection but do nothing
-                            nb_diff = 2
-                            break
-                    # one change has been done: use this new item to compare
-                    # with other
-                    if nb_diff <= 1:
-                        chg = True
-                        item1 = self._veclist[index1 - 1] = new_item
-                        index2 -= 1
-                        self._veclist.pop(index2)
-                    elif not full:
-                        # easy pass so break to avoid scanning all
-                        # index2; advance with next index1 for now
-                        break
-            if not chg and not full:
-                # if no change was done during the last normal pass, we do a
-                # full O(n^2) pass. This pass is done only at the end in the
-                # hope that most vectors have already been merged by easy
-                # O(n) passes.
-                chg = full = True
+            for pos in range(startpos, -1, -1):
+                if len(veclist) < 2:
+                    break
+                groups = {}
+                merged = False
+                for rgvec in veclist:
+                    key = tuple(map(frozenset, rgvec[:pos])) + \
+                          tuple(map(frozenset, rgvec[pos+1:]))
+                    gvec = groups.get(key)
+                    if gvec is None:
+                        groups[key] = rgvec
+                    else:
+                        # disjoint union on the only differing axis
+                        gvec[pos].update(rgvec[pos])
+                        merged = True
+                if merged:
+                    veclist = list(groups.values())
+                    chg = True
+            startpos = dim - 1
+        self._veclist = veclist
+        self._sort()
 
     def __or__(self, other):
         """Return the union of two RangeSetNDs as a new RangeSetND.

tests/NodeSetTest.py

@@ -2261,21 +2261,21 @@ def test_nd_nonoverlap(self):
         ns1.add("a[0-1]b[2-3]c[4-5]")
         self.assertEqual(ns1, NodeSet("a[0-1]b[2-3]c[4-5],a[0-2]b1c4,a2b[2-3]c4"))
         self.assertEqual(ns1, NodeSet("a2b[1-3]c4,a0b[1-2]c4,a0b3c[4-5],a1b[1-2]c4,a1b3c[4-5],a0b2c5,a1b2c5"))
-        self.assertEqual(str(ns1), "a[0-1]b[1-2]c4,a[0-1]b3c[4-5],a2b[1-3]c4,a[0-1]b2c5")
+        self.assertEqual(str(ns1), "a[0-1]b[2-3]c[4-5],a2b[1-3]c4,a[0-1]b1c4")
         self.assertEqual(len(ns1), 13)
 
         ns1 = NodeSet("a[0-1]b[2-3]c[4-5]")
         ns1.add("a[0-2]b[1-3]c[4]")
         self.assertEqual(ns1, NodeSet("a[0-1]b[2-3]c[4-5],a[0-2]b1c4,a2b[2-3]c4"))
         self.assertEqual(ns1, NodeSet("a2b[1-3]c4,a0b[1-2]c4,a0b3c[4-5],a1b[1-2]c4,a1b3c[4-5],a0b2c5,a1b2c5"))
-        self.assertEqual(str(ns1), "a[0-1]b[1-2]c4,a[0-1]b3c[4-5],a2b[1-3]c4,a[0-1]b2c5")
+        self.assertEqual(str(ns1), "a[0-1]b[2-3]c[4-5],a2b[1-3]c4,a[0-1]b1c4")
         self.assertEqual(len(ns1), 13)
 
         ns1 = NodeSet("a[0-2]b[1-3]c[4],a[0-1]b[2-3]c[4-5]")
         self.assertEqual(ns1, NodeSet("a[0-2]b[1-3]c[4],a[0-1]b[2-3]c[4-5]"))
         self.assertEqual(ns1, NodeSet("a2b[1-3]c4,a0b[1-2]c4,a0b3c[4-5],a1b[1-2]c4,a1b3c[4-5],a0b2c5,a1b2c5"))
         self.assertEqual(ns1, NodeSet("a[0-2]b[1-3]c4,a[0-1]b[2-3]c5"))
-        self.assertEqual(str(ns1), "a[0-1]b[1-2]c4,a[0-1]b3c[4-5],a2b[1-3]c4,a[0-1]b2c5")
+        self.assertEqual(str(ns1), "a[0-1]b[2-3]c[4-5],a2b[1-3]c4,a[0-1]b1c4")
         self.assertEqual(len(ns1), 13)
 
         ns1 = NodeSet("a[0-2]b[1-3]c[4-6],a[0-1]b[2-3]c[4-5]")
@@ -2330,7 +2330,7 @@ def test_nd_difference(self):
         self.assertEqual(len(ns1.difference(ns2)), 6)
 
         ns1 = NodeSet("a[0-2]b[1-3]c[4],a[0-1]b[2-3]c[4-5]")
-        self.assertEqual(str(ns1), "a[0-1]b[1-2]c4,a[0-1]b3c[4-5],a2b[1-3]c4,a[0-1]b2c5")
+        self.assertEqual(str(ns1), "a[0-1]b[2-3]c[4-5],a2b[1-3]c4,a[0-1]b1c4")
         self.assertEqual(ns1, NodeSet("a[0-2]b[1-3]c[4],a[0-1]b[2-3]c[4-5]"))
         self.assertEqual(ns1, NodeSet("a[0-1]b[2-3]c[4-5],a[0-2]b1c4,a2b[2-3]c4"))
         self.assertEqual(ns1, NodeSet("a2b[1-3]c4,a0b[1-2]c4,a0b3c[4-5],a1b[1-2]c4,a1b3c[4-5],a0b2c5,a1b2c5"))

tests/RangeSetNDTest.py

@@ -183,12 +183,41 @@ def test_folding(self):
             veclist.append((v1, v2, v3))
         self._testRS(veclist, "0; 5; 10\n1; 6; 11\n10; 15; 20\n11; 16; 21\n12; 17; 22\n13; 18; 23\n14; 19; 24\n15; 20; 25\n16; 21; 26\n17; 22; 27\n18; 23; 28\n19; 24; 29\n2; 7; 12\n20; 25; 30\n21; 26; 31\n22; 27; 32\n23; 28; 33\n24; 29; 34\n25; 30; 35\n26; 31; 36\n27; 32; 37\n28; 33; 38\n29; 34; 39\n3; 8; 13\n4; 9; 14\n5; 10; 15\n6; 11; 16\n7; 12; 17\n8; 13; 18\n9; 14; 19\n", 30)
 
+    def test_folding_multivariate(self):
+        # overlapping vectors absorbed into last-axis groups when expanding
+        self._testRS([["1", "2", "5"], ["1", "3", "5"], ["1", "2-3", "6"]],
+                     "1; 2-3; 5-6\n", 4)
+        # full 3D box rebuilt from single elements (multi-axis merges)
+        veclist = [(x, y, z) for x in (0, 1) for y in (0, 1) for z in (0, 1)]
+        self._testRS(veclist, "0-1; 0-1; 0-1\n", 8)
+        # merges cascading over both leading axes
+        self._testRS([["1", "1", "1"], ["1", "2", "1"], ["2", "1-2", "1"]],
+                     "1-2; 1-2; 1\n", 4)
+        # mixed length zero padding across overlapping vectors
+        self._testRS([["05", "1-2"], ["5", "1-2"], ["05-06", "2-3"]],
+                     "05; 1-3\n06; 2-3\n5; 1-2\n", 7)
+        # negative ranges
+        self._testRS([["-2-0", "1-2"], ["-1-1", "2-3"]],
+                     "-1-0; 1-3\n-2; 1-2\n1; 2-3\n", 10)
+
+    def test_folding_canonical(self):
+        # the same 13 nodes as two overlapping boxes, in both orders...
+        folded = "0-1; 2-3; 4-5\n2; 1-3; 4\n0-1; 1; 4\n"
+        self._testRS([["0-2", "1-3", "4"], ["0-1", "2-3", "4-5"]], folded, 13)
+        self._testRS([["0-1", "2-3", "4-5"], ["0-2", "1-3", "4"]], folded, 13)
+        # ...then as individual nodes in scrambled order
+        self._testRS([["1", "2", "5"], ["1", "2", "4"], ["0", "2", "5"],
+                      ["1", "3", "5"], ["2", "3", "4"], ["1", "1", "4"],
+                      ["2", "2", "4"], ["1", "3", "4"], ["0", "3", "4"],
+                      ["0", "3", "5"], ["0", "1", "4"], ["0", "2", "4"],
+                      ["2", "1", "4"]], folded, 13)
+
     def test_union(self):
         rn1 = RangeSetND([["10-100", "1-3"], ["1100-1300", "2-3"]])
         self.assertEqual(str(rn1), "1100-1300; 2-3\n10-100; 1-3\n")
         self.assertEqual(len(rn1), 675)
         rn2 = RangeSetND([["1100-1200", "1"], ["10-49", "1,3"]])
-        self.assertEqual(str(rn2), "12-13,1100-1200; 1\n10-11,14-49; 1,3\n12-13; 3\n")
+        self.assertEqual(str(rn2), "1100-1200; 1\n10-49; 1,3\n")
         self.assertEqual(len(rn2), 181)
         rnu = rn1.union(rn2)
         self.assertEqual(str(rnu), "10-100,1100-1200; 1-3\n1201-1300; 2-3\n")
@@ -209,7 +238,7 @@ def test_union(self):
             rn1 = RangeSetND([["10", "10-13"], ["10", "9-12"]])
             rn2 = RangeSetND([["1100-1200", "1"], ["10-49", "1,3"]])
             rn1 |= rn2
-            self.assertEqual(str(rn2), "12-13,1100-1200; 1\n10-11,14-49; 1,3\n12-13; 3\n")
+            self.assertEqual(str(rn2), "1100-1200; 1\n10-49; 1,3\n")
             self.assertEqual(len(rn2), 181)
             rn2 = set([3, 5])
             self.assertRaises(TypeError, rn1.__ior__, rn2)
@@ -575,3 +604,8 @@ def rebuilt(rgnd):
         rn.update(RangeSetND([["0-4,16", "0"]]))
         self.assertEqual(len(rn), 51)
         self.assertEqual(str(rn), "0-16; 0-2\n")
+        # difference_update() leaves axis objects shared between vectors
+        rn.difference_update(RangeSetND([["8-12", "0-2"]]))
+        self.assertEqual(len(rn), 36)
+        self.assertEqual(rn, rebuilt(rn))
+        self.assertEqual(str(rn), "0-7,13-16; 0-2\n")

tests/RangeSetTest.py

@@ -3,9 +3,12 @@
 
 """Unit test for RangeSet"""
 
+import ast
 import binascii
 import copy
+import inspect
 import pickle
+import sys
 import unittest
 import warnings
 
@@ -1486,6 +1489,36 @@ def test_set_mutators_are_overridden(self):
         self.assertEqual(missing, set(),
                          "set mutators not overridden: %s" % sorted(missing))
 
+    def test_no_raw_set_mutation_of_other_objects(self):
+        """test that inner sets are mutated through RangeSet methods only"""
+        # a raw set.<mutator>(x, ...) call skips the _sorted_cache reset:
+        # only allowed on self (inside the override) or on a freshly
+        # created, never-read set marked with a trailing '# fresh set'
+        mutators = set(vars(set)) - set(vars(frozenset))
+        mutators.discard('__getattribute__')  # not a mutator
+        source = inspect.getsource(sys.modules[RangeSet.__module__])
+        lines = source.splitlines()
+        offenders = []
+        for node in ast.walk(ast.parse(source)):
+            if not isinstance(node, ast.Call) or not node.args:
+                continue
+            func = node.func
+            if not isinstance(func, ast.Attribute) \
+                    or func.attr not in mutators:
+                continue
+            if not (isinstance(func.value, ast.Name)
+                    and func.value.id == 'set'):
+                continue
+            target = node.args[0]
+            if isinstance(target, ast.Name) and target.id == 'self':
+                continue
+            if lines[node.lineno - 1].endswith('# fresh set'):
+                continue
+            offenders.append("line %d: set.%s()" % (node.lineno, func.attr))
+        self.assertEqual(offenders, [],
+                         "raw set mutation of non-self object: %s"
+                         % ", ".join(offenders))
+
     def test_pop(self):
         """test RangeSet.pop()"""
         rset = RangeSet("1-3")