Repository: dask/dask

Issue:
`NestedContainer.to_container` method gets tracked individually per NestedContainer object
Looking into https://github.com/dask/distributed/issues/8958, I've noticed that for each `NestedContainer` object, its bound `to_container` method is tracked individually by the GC. This accounts for ~500k of 9MM objects in my workload. It would probably be better to stop tracking these individually.

Discussion:
github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     15 files  ±0       15 suites  ±0   4h 3m 19s ⏱️ + 2m 16s
 13 267 tests ±0   12 205 ✅ ±0   1 062 💤 ±0  0 ❌ ±0 
164 601 runs  ±0  141 674 ✅ ±0  22 927 💤 ±0  0 ❌ ±0 

Results for commit 9b47324a. ± Comparison against base commit 24c49209.

[test-results]:data:application/gzip;base64,H4sIAMaHYWcC/03MSw7CIBSF4a00jB3A5QLiZgwgJMQ+DIVR496ltaUd/t9JzkJC7P1MHh0Tt47MJeazXiWZHKdxbRRaV6pr3nYOUh39nItzKwJQceI7flakEpoFE/tqtIFPaUq7pDJu1xIlZXu3a2RSYcP/NYAGddjleuvrs5uGIeYaRFtUHNB4E4xHhxRAgWcWjOJMS8GcDxbvnHx/1IzVnhoBAAA=


Patch:
diff --git a/dask/_task_spec.py b/dask/_task_spec.py
index 316f1805a..438fdc3c6 100644
--- a/dask/_task_spec.py
+++ b/dask/_task_spec.py
@@ -84,7 +84,7 @@ _anom_count = itertools.count()
 def parse_input(obj: Any) -> object:
     """Tokenize user input into GraphNode objects
 
-    Note: This is similar to `convert_old_style_task` but does not
+    Note: This is similar to `convert_legacy_task` but does not
     - compare any values to a global set of known keys to infer references/futures
     - parse tuples and interprets them as runnable tasks
     - Deal with SubgraphCallables
@@ -101,20 +101,27 @@ def parse_input(obj: Any) -> object:
     """
     if isinstance(obj, GraphNode):
         return obj
-    if isinstance(obj, list):
-        return List(*(parse_input(o) for o in obj))
-    if isinstance(obj, set):
-        return Set(*(parse_input(o) for o in obj))
-    if isinstance(obj, tuple):
-        if is_namedtuple_instance(obj):
-            return _wrap_namedtuple_task(None, obj, parse_input)
-        return Tuple(*(parse_input(o) for o in obj))
-    if isinstance(obj, dict):
-        return Dict({k: parse_input(v) for k, v in obj.items()})
 
     if isinstance(obj, TaskRef):
         return Alias(obj.key)
 
+    if isinstance(obj, dict):
+        parsed_dict = {k: parse_input(v) for k, v in obj.items()}
+        if any(isinstance(v, GraphNode) for v in parsed_dict.values()):
+            return Dict(parsed_dict)
+
+    if isinstance(obj, (list, set, tuple)):
+        parsed_collection = tuple(parse_input(o) for o in obj)
+        if any(isinstance(o, GraphNode) for o in parsed_collection):
+            if isinstance(obj, list):
+                return List(*parsed_collection)
+            if isinstance(obj, set):
+                return Set(*parsed_collection)
+            if isinstance(obj, tuple):
+                if is_namedtuple_instance(obj):
+                    return _wrap_namedtuple_task(None, obj, parse_input)
+                return Tuple(*parsed_collection)
+
     return obj
 
 
@@ -799,6 +806,7 @@ class NestedContainer(Task):
             None,
             self.to_container,
             *args,
+            klass=self.klass,
             _dependencies=_dependencies,
             **kwargs,
         )
@@ -832,9 +840,9 @@ class NestedContainer(Task):
 
         return super().__dask_tokenize__()
 
-    @classmethod
-    def to_container(cls, *args, **kwargs):
-        return cls.klass(args)
+    @staticmethod
+    def to_container(*args, klass):
+        return klass(args)
 
 
 class List(NestedContainer):


Test patch:
diff --git a/dask/tests/test_task_spec.py b/dask/tests/test_task_spec.py
index 60885ec1f..c62447643 100644
--- a/dask/tests/test_task_spec.py
+++ b/dask/tests/test_task_spec.py
@@ -538,6 +538,11 @@ def test_pickle_literals():
     assert roundtripped == obj
 
 
+@pytest.mark.parametrize("obj", [set, {0}, [], [1], {}, {2: 3}, (), (4,)])
+def test_parse_non_task_inputs(obj):
+    assert parse_input(obj) == obj
+
+
 def test_resolve_aliases():
     tasks = [
         Alias("bar", "foo"),