fix(trace-explorer): Merge parallel spans (#70681)

Parallel spans can create a deeply nested timeline. This tries to merge
them together for a cleaner flatten visualization. We're explicitly not
merging root spans as that can create strange visualizations.

src/sentry/api/endpoints/organization_traces.py

@@ -41,6 +41,7 @@ class TraceInterval(TypedDict):
     kind: Literal["project", "missing", "other"]
     opCategory: str | None
     duration: int
+    isRoot: bool
     components: NotRequired[list[tuple[int, int]]]
 
 
@@ -722,7 +723,7 @@ def get_traces_breakdown_projects_query(
                 "precise.start_ts",
                 "precise.finish_ts",
             ],
-            orderby=["precise.start_ts", "precise.finish_ts"],
+            orderby=["precise.start_ts", "-precise.finish_ts"],
             # limit the number of segments we fetch per trace so a single
             # large trace does not result in the rest being blank
             limitby=("trace", int(MAX_SNUBA_RESULTS / len(trace_ids))),
@@ -760,10 +761,11 @@ def get_traces_breakdown_categories_query(
                 "transaction",
                 "span.category",
                 "sdk.name",
+                "parent_span",
                 "precise.start_ts",
                 "precise.finish_ts",
             ],
-            orderby=["precise.start_ts", "precise.finish_ts"],
+            orderby=["precise.start_ts", "-precise.finish_ts"],
             # limit the number of segments we fetch per trace so a single
             # large trace does not result in the rest being blank
             limitby=("trace", int(MAX_SNUBA_RESULTS / len(trace_ids))),
@@ -1000,7 +1002,12 @@ def process_breakdowns(data, traces_range):
 
     def should_merge(interval_a, interval_b):
         return (
-            interval_a["end"] >= interval_b["start"]
+            # only merge intervals that have parent spans, i.e. those that aren't the trace root
+            not interval_a["isRoot"]
+            and not interval_b["isRoot"]
+            # only merge intervals that overlap
+            and interval_a["end"] >= interval_b["start"]
+            # only merge intervals that are part of the same service
             and interval_a["project"] == interval_b["project"]
             and interval_a["sdkName"] == interval_b["sdkName"]
             and interval_a["opCategory"] == interval_b["opCategory"]
@@ -1032,14 +1039,16 @@ def breakdown_push(trace, interval):
                     "components": [
                         (last_interval["components"][-1][1], interval["components"][0][0]),
                     ],
+                    "isRoot": False,
                 }
             )
 
         breakdown.append(interval)
 
     def stack_push(trace, interval):
-        last_interval = stack_peek(trace)
-        if last_interval and should_merge(last_interval, interval):
+        for last_interval in reversed(stacks[trace]):
+            if not should_merge(last_interval, interval):
+                continue
             # update the end of this interval and it will
             # be updated in the breakdown as well
             last_interval["end"] = max(interval["end"], last_interval["end"])
@@ -1107,7 +1116,14 @@ def stack_clear(trace, until=None):
         row["quantized.start_ts"] = quantized_start
         row["quantized.finish_ts"] = quantized_end
 
-    data.sort(key=lambda row: (row["quantized.start_ts"], -row["quantized.finish_ts"]))
+    data.sort(
+        key=lambda row: (
+            row["quantized.start_ts"],
+            row["precise.start_ts"],
+            -row["quantized.finish_ts"],
+            -row["precise.finish_ts"],
+        )
+    )
 
     last_timestamp_per_trace: dict[str, int] = defaultdict(int)
 
@@ -1131,6 +1147,7 @@ def stack_clear(trace, until=None):
             "end": row["quantized.finish_ts"],
             "duration": 0,
             "components": [(row["precise.start_ts"], row["precise.finish_ts"])],
+            "isRoot": not bool(row.get("parent_span")),
         }
 
         # Clear the stack of any intervals that end before the current interval
@@ -1139,11 +1156,6 @@ def stack_clear(trace, until=None):
 
         stack_push(trace, cur)
 
-        # Clear the stack of any intervals that end before the current interval
-        # ends. Here we do not need to push them to the breakdowns because
-        # that time has already be attributed to the most recent interval.
-        stack_clear(trace, until=cur["end"])
-
     for trace, trace_range in traces_range.items():
         # Check to see if there is still a gap before the trace ends and fill it
         # with an other interval.
@@ -1158,6 +1170,7 @@ def stack_clear(trace, until=None):
             "start": other_start,
             "end": other_end,
             "duration": 0,
+            "isRoot": False,
         }
 
         # Clear the remaining intervals on the stack to find the latest end time