Merge pull request #33483 from rohitsinha54/fix-trie-merge

Fix bounded trie merge when merging on empty trie

sdks/python/apache_beam/metrics/cells.py

@@ -746,7 +746,7 @@ def merge(self, other: '_BoundedTrieNode') -> int:
       delta = 0
     elif not self._children:
       self._children = other._children
-      delta = self._size - other._size
+      delta = other._size - self._size
     else:
       delta = 0
       other_child: '_BoundedTrieNode'

sdks/python/apache_beam/metrics/cells_test.py

@@ -420,6 +420,23 @@ def test_bounded_trie_data_combine_trim(self):
             BoundedTrieData(root=right, bound=3)).get_result(),
         set([('a', True), ('b', 'd', False), ('c', 'd', False)]))
 
+  def test_merge_on_empty_node(self):
+    root1 = _BoundedTrieNode()
+    root2 = _BoundedTrieNode()
+    root2.add_all([["a", "b", "c"], ["a", "b", "d"], ["a", "e"]])
+    self.assertEqual(2, root1.merge(root2))
+    self.assertEqual(3, root1.size())
+    self.assertFalse(root1._truncated)
+
+  def test_merge_with_empty_node(self):
+    root1 = _BoundedTrieNode()
+    root1.add_all([["a", "b", "c"], ["a", "b", "d"], ["a", "e"]])
+    root2 = _BoundedTrieNode()
+
+    self.assertEqual(0, root1.merge(root2))
+    self.assertEqual(3, root1.size())
+    self.assertFalse(root1._truncated)
+
 
 if __name__ == '__main__':
   unittest.main()