Update

dask_expr/_core.py

@@ -117,7 +117,10 @@ def _tree_repr_lines(self, indent=0, recursive=True):
                         continue
                     op = f" branch_id={op.branch_id}"
                 header = self._tree_repr_argument_construction(i, op, header)
-
+        if self._branch_id.branch_id != 0:
+            header = self._tree_repr_argument_construction(
+                i + 1, f" branch_id={self._branch_id.branch_id}", header
+            )
         lines = [header] + lines
         lines = [" " * indent + line for line in lines]
 
@@ -604,7 +607,7 @@ def substitute_parameters(self, substitutions: dict) -> Expr:
             else:
                 new_operands.append(operand)
         if changed:
-            return type(self)(*new_operands)
+            return type(self)(*new_operands, _branch_id=self._branch_id)
         return self
 
     def _node_label_args(self):

dask_expr/io/io.py

@@ -105,7 +105,7 @@ def _name(self):
         return (
             funcname(type(self.operand("_expr"))).lower()
             + "-fused-"
-            + _tokenize_deterministic(*self.operands, self._branch_id)
+            + _tokenize_deterministic(*self.operands, self._expr._branch_id)
         )
 
     @functools.cached_property
@@ -416,7 +416,8 @@ def _simplify_up(self, parent, dependents):
                 return Literal(sum(_lengths))
 
         if isinstance(parent, Projection):
-            return super()._simplify_up(parent, dependents)
+            x = super()._simplify_up(parent, dependents)
+            return x
 
     def _divisions(self):
         return self._divisions_and_locations[0]