diff --git a/Android.bp b/Android.bp
index 7ab674c583..1bf21a8217 100644
--- a/Android.bp
+++ b/Android.bp
@@ -17823,6 +17823,7 @@ filegroup {
         "src/trace_processor/core/exec/column_view.cc",
         "src/trace_processor/core/exec/dataframe_scan.cc",
         "src/trace_processor/core/exec/group_by.cc",
+        "src/trace_processor/core/exec/interval_flatten.cc",
         "src/trace_processor/core/exec/interval_intersect.cc",
         "src/trace_processor/core/exec/key_encoder.cc",
         "src/trace_processor/core/exec/operator.cc",
@@ -17921,6 +17922,7 @@ filegroup {
     srcs: [
         "src/trace_processor/core/util/bit_vector_unittest.cc",
         "src/trace_processor/core/util/flex_vector_unittest.cc",
+        "src/trace_processor/core/util/heap_unittest.cc",
         "src/trace_processor/core/util/ops_unittest.cc",
         "src/trace_processor/core/util/slab_unittest.cc",
         "src/trace_processor/core/util/sort_unittest.cc",
diff --git a/BUILD b/BUILD
index 6cab5e4400..adcc6f0480 100644
--- a/BUILD
+++ b/BUILD
@@ -2587,6 +2587,8 @@ perfetto_filegroup(
         "src/trace_processor/core/exec/dataframe_scan.h",
         "src/trace_processor/core/exec/group_by.cc",
         "src/trace_processor/core/exec/group_by.h",
+        "src/trace_processor/core/exec/interval_flatten.cc",
+        "src/trace_processor/core/exec/interval_flatten.h",
         "src/trace_processor/core/exec/interval_intersect.cc",
         "src/trace_processor/core/exec/interval_intersect.h",
         "src/trace_processor/core/exec/key_encoder.cc",
@@ -2666,6 +2668,7 @@ perfetto_filegroup(
     srcs = [
         "src/trace_processor/core/util/bit_vector.h",
         "src/trace_processor/core/util/flex_vector.h",
+        "src/trace_processor/core/util/heap.h",
         "src/trace_processor/core/util/ops.cc",
         "src/trace_processor/core/util/ops.h",
         "src/trace_processor/core/util/range.h",
diff --git a/src/trace_processor/core/exec/BUILD.gn b/src/trace_processor/core/exec/BUILD.gn
index a50f3b01da..3c0fdded52 100644
--- a/src/trace_processor/core/exec/BUILD.gn
+++ b/src/trace_processor/core/exec/BUILD.gn
@@ -32,6 +32,8 @@ source_set("exec") {
     "dataframe_scan.h",
     "group_by.cc",
     "group_by.h",
+    "interval_flatten.cc",
+    "interval_flatten.h",
     "interval_intersect.cc",
     "interval_intersect.h",
     "key_encoder.cc",
@@ -117,6 +119,7 @@ if (enable_perfetto_benchmarks) {
     testonly = true
     sources = [
       "breaker_benchmark.cc",
+      "interval_flatten_benchmark.cc",
       "key_encoder_benchmark.cc",
       "tree_number_nodes_benchmark.cc",
       "tree_order_benchmark.cc",
diff --git a/src/trace_processor/core/exec/interval_flatten.cc b/src/trace_processor/core/exec/interval_flatten.cc
new file mode 100644
index 0000000000..bd113f19f0
--- /dev/null
+++ b/src/trace_processor/core/exec/interval_flatten.cc
@@ -0,0 +1,324 @@
+/*
+ * Copyright (C) 2026 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "src/trace_processor/core/exec/interval_flatten.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <limits>
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "perfetto/base/compiler.h"
+#include "perfetto/base/status.h"
+#include "perfetto/ext/base/small_vector.h"
+#include "src/trace_processor/core/common/storage_types.h"
+#include "src/trace_processor/core/exec/column_view.h"
+#include "src/trace_processor/core/util/heap.h"
+#include "src/trace_processor/core/util/span.h"
+
+namespace perfetto::trace_processor::core::exec {
+namespace {
+
+bool IsInt64(const RowBatch& batch, uint32_t index) {
+  const ColumnView& column = batch.column(index);
+  return column.kind() == ColumnView::Kind::kFlat && column.type().Is<Int64>();
+}
+
+constexpr auto kEndsLater = [](const auto& a, const auto& b) {
+  return a.end > b.end;
+};
+
+template <typename Live>
+void PushEnd(FlexVector<Live>& heap, Live live) {
+  heap.push_back(live);
+  HeapSiftUp(heap.data(), heap.size() - 1, live, kEndsLater);
+}
+
+template <typename Live>
+void PopEnd(FlexVector<Live>& heap) {
+  Live last = heap.back();
+  heap.pop_back();
+  if (!heap.empty()) {
+    HeapSiftDown(heap.data(), heap.size(), last, kEndsLater);
+  }
+}
+
+}  // namespace
+
+IntervalFlatten::IntervalFlatten(IntervalFlattenSpec spec)
+    : spec_(std::move(spec)) {
+  for (const IntervalFlattenSpec::Aggregate& agg : spec_.aggregates) {
+    sum_index_.push_back(sums_);
+    if (agg.function == IntervalFlattenSpec::Function::kSum) {
+      ++sums_;
+    }
+  }
+}
+
+IntervalFlatten::~IntervalFlatten() = default;
+IntervalFlatten::State::~State() = default;
+
+std::unique_ptr<Breaker::State> IntervalFlatten::CreateState() const {
+  auto state = std::make_unique<State>();
+  state->live.sums.resize(sums_);
+  state->instant = state->live;
+  state->segment_sums.resize(sums_);
+  return state;
+}
+
+bool IntervalFlatten::Consume(const RowBatch& in, Breaker::State& state) const {
+  auto& s = static_cast<State&>(state);
+  bool typed = IsInt64(in, spec_.ts_column) && IsInt64(in, spec_.dur_column);
+  for (const IntervalFlattenSpec::Aggregate& agg : spec_.aggregates) {
+    typed = typed && (agg.function != IntervalFlattenSpec::Function::kSum ||
+                      IsInt64(in, agg.column));
+  }
+  if (!typed) {
+    s.status = base::ErrStatus(
+        "INTERVAL FLATTEN: ts, dur and summed columns must be Int64");
+    return false;
+  }
+  FlatColumnReader<int64_t> ts(in.column(spec_.ts_column));
+  FlatColumnReader<int64_t> dur(in.column(spec_.dur_column));
+  base::SmallVector<FlatColumnReader<int64_t>, 4> sums;
+  for (const IntervalFlattenSpec::Aggregate& agg : spec_.aggregates) {
+    if (agg.function == IntervalFlattenSpec::Function::kSum) {
+      sums.emplace_back(in.column(agg.column));
+    }
+  }
+  bool keyed = !spec_.key_columns.empty();
+  const ColumnView* groups = keyed ? &in.column(spec_.group_column) : nullptr;
+  bool started = false;
+  uint32_t rows = in.size();
+  for (uint32_t row = 0; row < rows; ++row) {
+    int64_t start;
+    int64_t length;
+    if (!ts.Read(row, &start) || !dur.Read(row, &length)) {
+      continue;
+    }
+    if (start < 0 || length < 0) {
+      s.status =
+          base::ErrStatus("INTERVAL FLATTEN: a row's ts or dur is below zero");
+      return false;
+    }
+    uint32_t group = keyed ? groups->Value<uint32_t>(row) : 0;
+    if (!s.in_group || group != s.group) {
+      if (s.in_group) {
+        EndGroup(s);
+        ++s.group_number;
+      }
+      s.in_group = true;
+      s.group = group;
+      s.key_row = s.key_rows.size() + row;
+      s.time = std::numeric_limits<int64_t>::min();
+      started = true;
+    }
+    if (start != s.time) {
+      if (start < s.time) {
+        s.status = base::ErrStatus(
+            "INTERVAL FLATTEN: a group's rows must arrive in order of ts");
+        return false;
+      }
+      if (s.instant.count > 0) {
+        EmitInstant(s);
+      }
+      Advance(s, start);
+      s.time = start;
+    }
+    Totals* totals = &s.instant;
+    uint32_t slot = 0;
+    if (length > 0) {
+      totals = &s.live;
+      if (s.free_slots.empty()) {
+        slot = s.slots++;
+        s.slot_sums.resize(uint64_t{s.slots} * sums_);
+      } else {
+        slot = s.free_slots.back();
+        s.free_slots.pop_back();
+      }
+      PushEnd(s.ends, Live{start + length, slot});
+    }
+    ++totals->count;
+    for (uint32_t i = 0; i < sums_; ++i) {
+      Sum held{};
+      held.holding = sums[i].Read(row, &held.sum);
+      totals->sums[i].sum += held.sum;
+      totals->sums[i].holding += held.holding;
+      if (length > 0) {
+        s.slot_sums[size_t{slot} * sums_ + i] = held;
+      }
+    }
+  }
+  // Keys are read back from the row each group starts at.
+  if (!keyed || !started) {
+    return true;
+  }
+  s.retained.Reset();
+  for (uint32_t column : spec_.key_columns) {
+    s.retained.AddColumn(in.column(column), in.owner(column));
+  }
+  s.retained.SetCardinality(in.size());
+  s.status = s.key_rows.Append(s.retained);
+  return s.status.ok();
+}
+
+void IntervalFlatten::Advance(State& s, int64_t time) const {
+  int64_t previous = s.previous;
+  while (!s.ends.empty() && s.ends.begin()->end <= time) {
+    int64_t end = s.ends.begin()->end;
+    if (s.live.count > 0 && previous < end) {
+      Emit(s, previous, end - previous, false);
+    }
+    previous = end;
+    while (!s.ends.empty() && s.ends.begin()->end == end) {
+      uint32_t slot = s.ends.begin()->slot;
+      PopEnd(s.ends);
+      --s.live.count;
+      for (uint32_t i = 0; i < sums_; ++i) {
+        const Sum& held = s.slot_sums[size_t{slot} * sums_ + i];
+        s.live.sums[i].sum -= held.sum;
+        s.live.sums[i].holding -= held.holding;
+      }
+      s.free_slots.push_back(slot);
+    }
+  }
+  if (s.live.count > 0 && previous < time) {
+    Emit(s, previous, time - previous, false);
+  }
+  s.previous = time;
+}
+
+void IntervalFlatten::EmitInstant(State& s) const {
+  Emit(s, s.time, 0, true);
+  s.instant.count = 0;
+  std::fill(s.instant.sums.begin(), s.instant.sums.end(), Sum());
+}
+
+void IntervalFlatten::EndGroup(State& s) const {
+  if (s.instant.count > 0) {
+    EmitInstant(s);
+  }
+  Advance(s, std::numeric_limits<int64_t>::max());
+}
+
+void IntervalFlatten::Emit(State& s,
+                           int64_t ts,
+                           int64_t dur,
+                           bool with_instant) const {
+  uint32_t n = s.segments;
+  if (PERFETTO_UNLIKELY(n == s.segment_capacity)) {
+    GrowSegments(s);
+  }
+  s.segment_ts[n] = ts;
+  s.segment_dur[n] = dur;
+  s.segment_groups[n] = s.group_number;
+  s.segment_key_rows[n] = s.key_row;
+  const Totals& live = s.live;
+  const Totals& instant = s.instant;
+  s.segment_counts[n] = live.count + (with_instant ? instant.count : 0);
+  for (uint32_t i = 0; i < sums_; ++i) {
+    Sum total = live.sums[i];
+    if (with_instant) {
+      total.sum += instant.sums[i].sum;
+      total.holding += instant.sums[i].holding;
+    }
+    State::SegmentSums& out = s.segment_sums[i];
+    out.values[n] = total.sum;
+    out.present.change(n, total.holding > 0);
+  }
+  s.segments = n + 1;
+}
+
+PERFETTO_NO_INLINE void IntervalFlatten::GrowSegments(State& s) const {
+  s.segment_capacity = std::max(64u, s.segment_capacity * 2);
+  s.segment_ts.resize(s.segment_capacity);
+  s.segment_dur.resize(s.segment_capacity);
+  s.segment_groups.resize(s.segment_capacity);
+  s.segment_key_rows.resize(s.segment_capacity);
+  s.segment_counts.resize(s.segment_capacity);
+  for (State::SegmentSums& sum : s.segment_sums) {
+    sum.values.resize(s.segment_capacity);
+    sum.present.resize(s.segment_capacity);
+  }
+}
+
+bool IntervalFlatten::Finalize(Breaker::State& state) const {
+  auto& s = static_cast<State&>(state);
+  if (s.in_group) {
+    EndGroup(s);
+  }
+  return true;
+}
+
+bool IntervalFlatten::Serve(RowBatch& out, Breaker::State& state) const {
+  auto& s = static_cast<State&>(state);
+  uint32_t count = std::min(s.segments - s.served, kMaxBatchRows);
+  if (count == 0) {
+    return false;
+  }
+  auto add = [&](StorageType type, const void* values,
+                 const BitVector* validity) {
+    ColumnView view = ColumnView::Reference(type, values, validity);
+    view.SetRange(s.served);
+    out.AddColumn(view);
+  };
+  add(StorageType{Int64{}}, s.segment_ts.data(), nullptr);
+  add(StorageType{Int64{}}, s.segment_dur.data(), nullptr);
+  if (!spec_.key_columns.empty()) {
+    const uint32_t* rows = s.segment_key_rows.data() + s.served;
+    s.key_rows.View(&s.served_keys, {rows, rows + count});
+    for (uint32_t k = 0; k < s.served_keys.column_count(); ++k) {
+      out.AddColumn(s.served_keys.column(k));
+    }
+  }
+  for (uint32_t a = 0; a < spec_.aggregates.size(); ++a) {
+    if (spec_.aggregates[a].function == IntervalFlattenSpec::Function::kSum) {
+      const State::SegmentSums& sum = s.segment_sums[sum_index_[a]];
+      add(StorageType{Int64{}}, sum.values.data(), &sum.present);
+    } else {
+      add(StorageType{Int64{}}, s.segment_counts.data(), nullptr);
+    }
+  }
+  add(StorageType{Uint32{}}, s.segment_groups.data(), nullptr);
+  out.SetCardinality(count);
+  s.served += count;
+  return true;
+}
+
+void IntervalFlatten::Reset(Breaker::State& state) const {
+  auto& s = static_cast<State&>(state);
+  s.group = 0;
+  s.group_number = 0;
+  s.in_group = false;
+  s.previous = 0;
+  s.time = 0;
+  s.live.count = 0;
+  std::fill(s.live.sums.begin(), s.live.sums.end(), Sum());
+  s.instant.count = 0;
+  std::fill(s.instant.sums.begin(), s.instant.sums.end(), Sum());
+  s.ends.clear();
+  s.slots = 0;
+  s.slot_sums.clear();
+  s.free_slots.clear();
+  s.key_rows.Clear();
+  s.segments = 0;
+  s.key_row = 0;
+  s.served = 0;
+}
+
+}  // namespace perfetto::trace_processor::core::exec
diff --git a/src/trace_processor/core/exec/interval_flatten.h b/src/trace_processor/core/exec/interval_flatten.h
new file mode 100644
index 0000000000..d1be8f35af
--- /dev/null
+++ b/src/trace_processor/core/exec/interval_flatten.h
@@ -0,0 +1,134 @@
+/*
+ * Copyright (C) 2026 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef SRC_TRACE_PROCESSOR_CORE_EXEC_INTERVAL_FLATTEN_H_
+#define SRC_TRACE_PROCESSOR_CORE_EXEC_INTERVAL_FLATTEN_H_
+
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+#include "src/trace_processor/core/exec/breaker.h"
+#include "src/trace_processor/core/exec/operator.h"
+#include "src/trace_processor/core/exec/row_batch.h"
+#include "src/trace_processor/core/exec/row_store.h"
+#include "src/trace_processor/core/util/bit_vector.h"
+#include "src/trace_processor/core/util/flex_vector.h"
+
+namespace perfetto::trace_processor::core::exec {
+
+struct IntervalFlattenSpec {
+  enum class Function : uint8_t { kCount, kSum };
+  struct Aggregate {
+    Function function = Function::kCount;
+    // Unused by kCount.
+    uint32_t column = 0;
+  };
+  uint32_t ts_column = 0;
+  uint32_t dur_column = 0;
+  std::vector<uint32_t> key_columns;
+  // As GroupBy appends. Unused without keys.
+  uint32_t group_column = 0;
+  std::vector<Aggregate> aggregates;
+};
+
+// INTERVAL FLATTEN. The input must be grouped by the keys and ordered by ts
+// within each group; the output is too, followed by a group number column.
+// A row of no width becomes a segment of no width which also counts the rows
+// spanning it.
+class IntervalFlatten : public Breaker {
+ public:
+  explicit IntervalFlatten(IntervalFlattenSpec);
+  ~IntervalFlatten() override;
+  BatchPreference batch_preference() const override {
+    return BatchPreference::kThroughput;
+  }
+
+ private:
+  // A sum over some rows, and how many of them held a value.
+  struct Sum {
+    int64_t sum;
+    int64_t holding;
+  };
+  struct Totals {
+    int64_t count = 0;
+    std::vector<Sum> sums;
+  };
+  struct Live {
+    int64_t end;
+    uint32_t slot;
+  };
+
+  struct State : Breaker::State {
+    ~State() override;
+
+    uint32_t group = 0;
+    uint32_t group_number = 0;
+    bool in_group = false;
+    int64_t previous = 0;
+    int64_t time = 0;
+    Totals live;
+    Totals instant;
+    // A min-heap on end.
+    FlexVector<Live> ends;
+    // Sum values of live rows, by slot.
+    uint32_t slots = 0;
+    FlexVector<Sum> slot_sums;
+    FlexVector<uint32_t> free_slots;
+
+    RowStore key_rows;
+    RowBatch retained;
+
+    // The segments, a column each. All are `segment_capacity` long, of which
+    // the first `segments` are used.
+    uint32_t segments = 0;
+    uint32_t segment_capacity = 0;
+    FlexVector<int64_t> segment_ts;
+    FlexVector<int64_t> segment_dur;
+    FlexVector<uint32_t> segment_groups;
+    FlexVector<uint32_t> segment_key_rows;
+    uint32_t key_row = 0;
+    // What every kCount aggregate holds.
+    FlexVector<int64_t> segment_counts;
+    struct SegmentSums {
+      FlexVector<int64_t> values;
+      BitVector present;
+    };
+    std::vector<SegmentSums> segment_sums;
+    uint32_t served = 0;
+    RowBatch served_keys;
+  };
+
+  std::unique_ptr<Breaker::State> CreateState() const override;
+  bool Consume(const RowBatch& in, Breaker::State& state) const override;
+  bool Finalize(Breaker::State& state) const override;
+  bool Serve(RowBatch& out, Breaker::State& state) const override;
+  void Reset(Breaker::State& state) const override;
+
+  void Advance(State&, int64_t time) const;
+  void EmitInstant(State&) const;
+  void EndGroup(State&) const;
+  void Emit(State&, int64_t ts, int64_t dur, bool with_instant) const;
+  void GrowSegments(State&) const;
+
+  IntervalFlattenSpec spec_;
+  std::vector<uint32_t> sum_index_;
+  uint32_t sums_ = 0;
+};
+
+}  // namespace perfetto::trace_processor::core::exec
+
+#endif  // SRC_TRACE_PROCESSOR_CORE_EXEC_INTERVAL_FLATTEN_H_
diff --git a/src/trace_processor/core/exec/interval_flatten_benchmark.cc b/src/trace_processor/core/exec/interval_flatten_benchmark.cc
new file mode 100644
index 0000000000..2f25b8e95e
--- /dev/null
+++ b/src/trace_processor/core/exec/interval_flatten_benchmark.cc
@@ -0,0 +1,90 @@
+/*
+ * Copyright (C) 2026 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <benchmark/benchmark.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+#include "src/trace_processor/core/common/storage_types.h"
+#include "src/trace_processor/core/exec/column_view.h"
+#include "src/trace_processor/core/exec/interval_flatten.h"
+#include "src/trace_processor/core/exec/operator.h"
+#include "src/trace_processor/core/exec/row_batch.h"
+#include "src/trace_processor/core/exec/row_selection.h"
+
+// Flattening intervals about five deep, counting and summing them: over 16
+// full batches, as over a large input, and over one row, as when a small input
+// is run over and over.
+
+namespace perfetto::trace_processor::core::exec {
+namespace {
+
+void Run(benchmark::State& state, uint32_t rows) {
+  std::vector<int64_t> ts;
+  std::vector<int64_t> dur;
+  std::vector<int64_t> value;
+  for (uint32_t i = 0; i < rows; ++i) {
+    uint32_t hash = i * 2654435761u;
+    ts.push_back(int64_t{i} * 10);
+    dur.push_back(hash >> 26);
+    value.push_back(hash >> 24);
+  }
+  std::vector<RowBatch> batches;
+  for (uint32_t at = 0; at < rows; at += kMaxBatchRows) {
+    uint32_t count = std::min(kMaxBatchRows, rows - at);
+    RowBatch& batch = batches.emplace_back();
+    for (const std::vector<int64_t>* column : {&ts, &dur, &value}) {
+      batch.AddColumn(
+          ColumnView::Reference(StorageType{Int64{}}, column->data()));
+    }
+    batch.Compose(RowSelection::Range(at), count);
+    batch.SetCardinality(count);
+  }
+  IntervalFlattenSpec spec;
+  spec.ts_column = 0;
+  spec.dur_column = 1;
+  spec.aggregates = {{IntervalFlattenSpec::Function::kCount, 0},
+                     {IntervalFlattenSpec::Function::kSum, 2}};
+  IntervalFlatten op(spec);
+  std::unique_ptr<OperatorState> op_state = op.MakeState();
+  RowBatch out;
+  for (auto _ : state) {
+    for (const RowBatch& batch : batches) {
+      op.Execute(batch, out, *op_state);
+    }
+    while (op.Finish(out, *op_state) == OpResult::kHaveMoreOutput) {
+      benchmark::DoNotOptimize(out.size());
+    }
+    op.Rewind(*op_state);
+  }
+  state.SetItemsProcessed(static_cast<int64_t>(state.iterations()) * rows);
+}
+
+void BM_IntervalFlatten(benchmark::State& state) {
+  Run(state, 16 * kMaxBatchRows);
+}
+BENCHMARK(BM_IntervalFlatten);
+
+void BM_IntervalFlattenOneRow(benchmark::State& state) {
+  Run(state, 1);
+}
+BENCHMARK(BM_IntervalFlattenOneRow);
+
+}  // namespace
+}  // namespace perfetto::trace_processor::core::exec
diff --git a/src/trace_processor/core/util/BUILD.gn b/src/trace_processor/core/util/BUILD.gn
index 6077b81423..cd0adfdadc 100644
--- a/src/trace_processor/core/util/BUILD.gn
+++ b/src/trace_processor/core/util/BUILD.gn
@@ -18,6 +18,7 @@ source_set("util") {
   sources = [
     "bit_vector.h",
     "flex_vector.h",
+    "heap.h",
     "ops.cc",
     "ops.h",
     "range.h",
@@ -38,6 +39,7 @@ perfetto_unittest_source_set("unittests") {
   sources = [
     "bit_vector_unittest.cc",
     "flex_vector_unittest.cc",
+    "heap_unittest.cc",
     "ops_unittest.cc",
     "slab_unittest.cc",
     "sort_unittest.cc",
diff --git a/src/trace_processor/core/util/heap.h b/src/trace_processor/core/util/heap.h
new file mode 100644
index 0000000000..881acc50b5
--- /dev/null
+++ b/src/trace_processor/core/util/heap.h
@@ -0,0 +1,73 @@
+/*
+ * Copyright (C) 2026 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef SRC_TRACE_PROCESSOR_CORE_UTIL_HEAP_H_
+#define SRC_TRACE_PROCESSOR_CORE_UTIL_HEAP_H_
+
+#include <cstddef>
+
+#include "perfetto/base/compiler.h"
+
+// A binary heap in an array, with the element which is not `later` than any
+// other on top: the same layout as std::make_heap() with `later` as its
+// comparator. Unlike std::push_heap() and std::pop_heap(), these place a value
+// in one pass which moves the elements it passes over, and do not swap it
+// through a temporary at each level.
+
+namespace perfetto::trace_processor::core {
+
+// Places `value` in `heap[0, size)`, whose top is free: for replacing the top,
+// or for removing it with the last element as `value`.
+template <typename T, typename Later>
+PERFETTO_ALWAYS_INLINE void HeapSiftDown(T* heap,
+                                         size_t size,
+                                         T value,
+                                         const Later& later) {
+  size_t at = 0;
+  for (size_t child = 1; child < size; child = (2 * at) + 1) {
+    if (child + 1 < size && later(heap[child], heap[child + 1])) {
+      ++child;
+    }
+    if (!later(value, heap[child])) {
+      break;
+    }
+    heap[at] = heap[child];
+    at = child;
+  }
+  heap[at] = value;
+}
+
+// Places `value` in the heap `heap[0, at]`, whose last element `heap[at]` is
+// free: for adding one.
+template <typename T, typename Later>
+PERFETTO_ALWAYS_INLINE void HeapSiftUp(T* heap,
+                                       size_t at,
+                                       T value,
+                                       const Later& later) {
+  while (at > 0) {
+    size_t parent = (at - 1) / 2;
+    if (!later(heap[parent], value)) {
+      break;
+    }
+    heap[at] = heap[parent];
+    at = parent;
+  }
+  heap[at] = value;
+}
+
+}  // namespace perfetto::trace_processor::core
+
+#endif  // SRC_TRACE_PROCESSOR_CORE_UTIL_HEAP_H_
diff --git a/src/trace_processor/core/util/heap_unittest.cc b/src/trace_processor/core/util/heap_unittest.cc
new file mode 100644
index 0000000000..e86d0e1f76
--- /dev/null
+++ b/src/trace_processor/core/util/heap_unittest.cc
@@ -0,0 +1,76 @@
+/*
+ * Copyright (C) 2026 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "src/trace_processor/core/util/heap.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <random>
+#include <vector>
+
+#include "test/gtest_and_gmock.h"
+
+namespace perfetto::trace_processor::core {
+namespace {
+
+constexpr auto kLater = [](uint32_t a, uint32_t b) { return a > b; };
+
+void Push(std::vector<uint32_t>& heap, uint32_t value) {
+  heap.push_back(value);
+  HeapSiftUp(heap.data(), heap.size() - 1, value, kLater);
+}
+
+uint32_t Pop(std::vector<uint32_t>& heap) {
+  uint32_t top = heap[0];
+  uint32_t last = heap.back();
+  heap.pop_back();
+  if (!heap.empty()) {
+    HeapSiftDown(heap.data(), heap.size(), last, kLater);
+  }
+  return top;
+}
+
+// Pushes, pops and replaces the top at random, against a sorted multiset,
+// checking that each pop is the smallest and that the layout stays a heap as
+// std::is_heap() sees it.
+TEST(HeapTest, MatchesSortedOrder) {
+  std::minstd_rand0 rnd(0);
+  std::vector<uint32_t> heap;
+  std::vector<uint32_t> expected;
+  for (uint32_t i = 0; i < 20000; ++i) {
+    uint32_t op = rnd() % 3;
+    if (op == 0 || expected.empty()) {
+      uint32_t value = rnd() % 100;
+      Push(heap, value);
+      expected.insert(std::upper_bound(expected.begin(), expected.end(), value),
+                      value);
+    } else if (op == 1) {
+      ASSERT_EQ(Pop(heap), expected.front());
+      expected.erase(expected.begin());
+    } else {
+      // Replace the top with a later value, as a merge does.
+      uint32_t value = heap[0] + rnd() % 10;
+      HeapSiftDown(heap.data(), heap.size(), value, kLater);
+      expected.erase(expected.begin());
+      expected.insert(std::upper_bound(expected.begin(), expected.end(), value),
+                      value);
+    }
+    ASSERT_TRUE(std::is_heap(heap.begin(), heap.end(), kLater));
+  }
+}
+
+}  // namespace
+}  // namespace perfetto::trace_processor::core
diff --git a/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection_unittest.cc b/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection_unittest.cc
index d09e021f9e..b7d58cfb0a 100644
--- a/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection_unittest.cc
+++ b/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection_unittest.cc
@@ -24,6 +24,7 @@
 
 #include "perfetto/ext/base/status_macros.h"
 #include "perfetto/ext/base/status_or.h"
+#include "perfetto/ext/base/string_utils.h"
 #include "src/base/test/status_matchers.h"
 #include "src/trace_processor/containers/string_pool.h"
 #include "src/trace_processor/sqlite/bindings/sqlite_result.h"
@@ -1202,12 +1203,90 @@ TEST_F(PerfettoSqlConnectionPipelineTest, IntervalIntersectionPerAnyType) {
               testing::HasSubstr("the same in every operand"));
 }
 
-TEST_F(PerfettoSqlConnectionPipelineTest, IntervalFlattenParses) {
-  EXPECT_THAT(Rows("FROM (SELECT 0 AS ts, 1 AS dur) "
+TEST_F(PerfettoSqlConnectionPipelineTest, IntervalFlatten) {
+  ASSERT_TRUE(Rows(R"(
+    CREATE TABLE spans(ts INTEGER, dur INTEGER, cpu INTEGER, weight INTEGER);
+    INSERT INTO spans VALUES
+      (0, 10, 1, 5), (5, 10, 1, NULL), (15, 5, 1, 2), (7, 0, 1, 1),
+      (30, 5, 2, 4), (40, 5, NULL, 1), (NULL, 5, 1, 1);
+  )")
+                  .ok());
+  // Overlaps are cut at every boundary; a point makes an instant counting the
+  // rows spanning it; rows which only meet stay apart; nulls sum to nothing.
+  auto rows = Rows(R"(
+    FROM spans
+    |> INTERVAL FLATTEN PER cpu
+       AGGREGATE COUNT(*) AS n, SUM(weight) AS w
+  )");
+  ASSERT_TRUE(rows.ok()) << rows.status().message();
+  EXPECT_THAT(*rows,
+              testing::ElementsAre("0,5,1,1,5", "10,5,1,1,NULL", "15,5,1,1,2",
+                                   "30,5,2,1,4", "40,5,NULL,1,1", "5,2,1,2,5",
+                                   "7,0,1,3,6", "7,3,1,2,5"));
+
+  rows = Rows(R"(
+    FROM (SELECT ts, dur FROM spans WHERE cpu = 2 OR cpu IS NULL)
+    |> INTERVAL FLATTEN AGGREGATE COUNT(*) AS n
+    |> SELECT ts, n
+  )");
+  ASSERT_TRUE(rows.ok()) << rows.status().message();
+  EXPECT_THAT(*rows, testing::ElementsAre("30,1", "40,1"));
+
+  // Keys can be of any one type, and come out as they went in.
+  rows = Rows(R"(
+    FROM (
+      SELECT ts, dur, IIF(cpu = 1, 'one', NULL) AS c FROM spans
+    )
+    |> INTERVAL FLATTEN PER c AGGREGATE COUNT(*) AS n
+  )");
+  ASSERT_TRUE(rows.ok()) << rows.status().message();
+  EXPECT_THAT(*rows, testing::ElementsAre(
+                         "0,5,one,1", "10,5,one,1", "15,5,one,1", "30,5,NULL,1",
+                         "40,5,NULL,1", "5,2,one,2", "7,0,one,3", "7,3,one,2"));
+
+  // Input order doesn't matter.
+  for (const char* order : {"ts", "ts DESC"}) {
+    rows = Rows(base::StackString<256>(R"(
+      FROM (SELECT * FROM spans ORDER BY %s)
+      |> INTERVAL FLATTEN PER cpu
+         AGGREGATE COUNT(*) AS n, SUM(weight) AS w
+    )",
+                                       order)
+                    .ToStdString());
+    ASSERT_TRUE(rows.ok()) << rows.status().message();
+    EXPECT_THAT(*rows,
+                testing::ElementsAre("0,5,1,1,5", "10,5,1,1,NULL", "15,5,1,1,2",
+                                     "30,5,2,1,4", "40,5,NULL,1,1", "5,2,1,2,5",
+                                     "7,0,1,3,6", "7,3,1,2,5"));
+  }
+
+  // Segments never overlap, so flattening again keeps each.
+  rows = Rows(R"(
+    FROM (SELECT * FROM spans WHERE dur > 0)
+    |> INTERVAL FLATTEN PER cpu AGGREGATE COUNT(*) AS n
+    |> INTERVAL FLATTEN PER cpu AGGREGATE COUNT(*) AS m, SUM(n) AS n
+  )");
+  ASSERT_TRUE(rows.ok()) << rows.status().message();
+  EXPECT_THAT(*rows,
+              testing::ElementsAre("0,5,1,1,1", "10,5,1,1,1", "15,5,1,1,1",
+                                   "30,5,2,1,1", "40,5,NULL,1,1", "5,5,1,1,2"));
+
+  // Only the segment's columns are left.
+  EXPECT_THAT(Rows("FROM spans |> INTERVAL FLATTEN AGGREGATE COUNT(*) AS n "
+                   "|> SELECT weight")
+                  .status()
+                  .message(),
+              testing::HasSubstr("weight"));
+  EXPECT_THAT(Rows("FROM (SELECT 0 AS ts, -1 AS dur) "
                    "|> INTERVAL FLATTEN AGGREGATE COUNT(*) AS n")
                   .status()
                   .message(),
-              testing::HasSubstr("INTERVAL FLATTEN is not supported yet"));
+              testing::HasSubstr("below zero"));
+  EXPECT_THAT(Rows("FROM spans |> INTERVAL FLATTEN "
+                   "AGGREGATE COUNT(weight) AS n")
+                  .status()
+                  .message(),
+              testing::HasSubstr("COUNT"));
 }
 
 TEST_F(PerfettoSqlConnectionPipelineTest, ForksRunPipelinesIndependently) {
diff --git a/src/trace_processor/perfetto_sql/pipeline/column_pruning.cc b/src/trace_processor/perfetto_sql/pipeline/column_pruning.cc
index 24a71844ea..8d8f95c6de 100644
--- a/src/trace_processor/perfetto_sql/pipeline/column_pruning.cc
+++ b/src/trace_processor/perfetto_sql/pipeline/column_pruning.cc
@@ -134,6 +134,33 @@ PlanNodeId PruneIntervalIntersect(LogicalPlan& plan,
   return id;
 }
 
+PlanNodeId PruneIntervalFlatten(LogicalPlan& plan,
+                                PlanNodeId id,
+                                Needed& needed) {
+  PlanNode& node = plan.nodes[id];
+  auto& flatten = node.Cast<op::IntervalFlatten>();
+  auto& aggregates = flatten.aggregates;
+  aggregates.erase(
+      std::remove_if(aggregates.begin(), aggregates.end(),
+                     [&](const op::IntervalFlatten::Aggregate& agg) {
+                       return !needed[agg.output];
+                     }),
+      aggregates.end());
+  // Unlike a fold, it stays with no aggregates left: its rows are segments.
+  needed[flatten.ts] = true;
+  needed[flatten.dur] = true;
+  for (ColumnId key : flatten.keys) {
+    needed[key] = true;
+  }
+  for (const op::IntervalFlatten::Aggregate& agg : aggregates) {
+    if (agg.function == op::IntervalFlatten::Function::kSum) {
+      needed[agg.column] = true;
+    }
+  }
+  node.children[0] = PruneNode(plan, node.children[0], needed);
+  return id;
+}
+
 PlanNodeId PruneNode(LogicalPlan& plan, PlanNodeId id, Needed& needed) {
   PlanNode& node = plan.nodes[id];
   switch (node.op.index()) {
@@ -144,6 +171,8 @@ PlanNodeId PruneNode(LogicalPlan& plan, PlanNodeId id, Needed& needed) {
       return PruneTreeAccumulate(plan, id, needed);
     case base::variant_index<Op, op::IntervalIntersect>():
       return PruneIntervalIntersect(plan, id, needed);
+    case base::variant_index<Op, op::IntervalFlatten>():
+      return PruneIntervalFlatten(plan, id, needed);
     default:
       PERFETTO_FATAL("Unknown operator");
   }
diff --git a/src/trace_processor/perfetto_sql/pipeline/compiler.cc b/src/trace_processor/perfetto_sql/pipeline/compiler.cc
index 87ba5c9461..ccbfd3d401 100644
--- a/src/trace_processor/perfetto_sql/pipeline/compiler.cc
+++ b/src/trace_processor/perfetto_sql/pipeline/compiler.cc
@@ -164,6 +164,8 @@ class Compiler {
   base::StatusOr<op::Scan> CompileSqlSource(uint32_t from);
   void AddScanColumn(op::Scan&, ColumnSchema);
   base::Status CompileTreeAccumulate(uint32_t stage);
+  base::Status CompileIntervalFlatten(uint32_t stage);
+  bool IsCountStar(uint32_t expr) const;
   base::StatusOr<ColumnId> CompileSum(uint32_t agg_id, uint32_t expr);
   base::StatusOr<ColumnId> Resolve(const std::string& name, uint32_t at) const {
     return Resolve("", name, at);
@@ -438,6 +440,8 @@ base::Status Compiler::CompileStage(uint32_t stage) {
   switch (static_cast<int>(node->tag)) {
     case SYNTAQLITE_NODE_PERFETTO_TREE_ACCUMULATE:
       return CompileTreeAccumulate(stage);
+    case SYNTAQLITE_NODE_PERFETTO_INTERVAL_FLATTEN:
+      return CompileIntervalFlatten(stage);
     case SYNTAQLITE_NODE_PERFETTO_PIPE_SELECT:
       return CompileSelect(stage);
     case SYNTAQLITE_NODE_PERFETTO_PIPE_EXTEND:
@@ -450,8 +454,6 @@ base::Status Compiler::CompileStage(uint32_t stage) {
       return CompileSet(stage);
     case SYNTAQLITE_NODE_PERFETTO_PIPE_AS:
       return CompileAs(stage);
-    case SYNTAQLITE_NODE_PERFETTO_INTERVAL_FLATTEN:
-      return Unsupported(stage, "INTERVAL FLATTEN");
     default:
       PERFETTO_FATAL("Unknown pipeline stage");
   }
@@ -675,6 +677,77 @@ base::Status Compiler::CompileTreeAccumulate(uint32_t stage) {
   return base::OkStatus();
 }
 
+bool Compiler::IsCountStar(uint32_t expr) const {
+  const auto* node = Node<SyntaqliteNode>(p_, expr);
+  if (node->tag != SYNTAQLITE_NODE_FUNCTION_CALL) {
+    return false;
+  }
+  const SyntaqliteFunctionCall& call = node->function_call;
+  return call.flags.bits.star && !call.flags.bits.distinct &&
+         !syntaqlite_node_is_present(call.filter_clause) &&
+         !syntaqlite_node_is_present(call.over_clause) &&
+         base::CaseInsensitiveEqual(SpanText(p_, call.func_name), "COUNT");
+}
+
+base::Status Compiler::CompileIntervalFlatten(uint32_t stage) {
+  scope_.op = "INTERVAL FLATTEN";
+  const auto* n = Node<SyntaqlitePerfettoIntervalFlatten>(p_, stage);
+
+  op::IntervalFlatten flatten;
+  ASSIGN_OR_RETURN(flatten.ts, Resolve("ts", stage));
+  ASSIGN_OR_RETURN(flatten.dur, Resolve("dur", stage));
+
+  std::vector<std::pair<NamedColumn, uint32_t>> keys;
+  if (syntaqlite_node_is_present(n->per)) {
+    const auto* per = Node<SyntaqlitePerfettoPerColumnList>(p_, n->per);
+    std::vector<std::string> seen;
+    for (uint32_t i = 0; i < syntaqlite_list_count(per); ++i) {
+      uint32_t col_id = syntaqlite_list_child_id(per, i);
+      std::string name =
+          SpanText(p_, Node<SyntaqlitePerfettoPerColumn>(p_, col_id)->name);
+      RETURN_IF_ERROR(CheckListedOnce(seen, name, col_id));
+      ASSIGN_OR_RETURN(ColumnId key, Resolve(name, col_id));
+      flatten.keys.push_back(key);
+      keys.push_back({NamedColumn{std::move(name), key}, col_id});
+    }
+  }
+
+  std::vector<std::pair<NamedColumn, uint32_t>> aggregates;
+  const auto* list = Node<SyntaqlitePerfettoAggregateList>(p_, n->aggregates);
+  for (uint32_t i = 0; i < syntaqlite_list_count(list); ++i) {
+    uint32_t agg_id = syntaqlite_list_child_id(list, i);
+    const auto* agg = Node<SyntaqlitePerfettoAggregate>(p_, agg_id);
+    op::IntervalFlatten::Aggregate aggregate;
+    if (IsCountStar(agg->expr)) {
+      aggregate.function = op::IntervalFlatten::Function::kCount;
+    } else {
+      aggregate.function = op::IntervalFlatten::Function::kSum;
+      ASSIGN_OR_RETURN(aggregate.column, CompileSum(agg_id, agg->expr));
+    }
+    std::string name = SpanText(p_, agg->name);
+    aggregate.output = plan_.AddColumn(name, core::Int64{});
+    flatten.aggregates.push_back(aggregate);
+    aggregates.push_back(
+        {NamedColumn{std::move(name), aggregate.output}, agg_id});
+  }
+
+  // The rows collapse, so the row is only what each segment holds.
+  flatten.out_ts = plan_.AddColumn("ts", core::Int64{});
+  flatten.out_dur = plan_.AddColumn("dur", core::Int64{});
+  scope_.row.clear();
+  scope_.aliases.clear();
+  Append({"ts", flatten.out_ts}, stage);
+  Append({"dur", flatten.out_dur}, stage);
+  for (auto& [column, node] : keys) {
+    Append(std::move(column), node);
+  }
+  for (auto& [column, node] : aggregates) {
+    Append(std::move(column), node);
+  }
+  plan_.AddNode(std::move(flatten), {plan_.root});
+  return base::OkStatus();
+}
+
 // The stages below are relational operators which only change which columns
 // the row has and what they are called: nothing runs. Where an operator could
 // take an expression, it takes a column reference for now.
diff --git a/src/trace_processor/perfetto_sql/pipeline/logical_plan.h b/src/trace_processor/perfetto_sql/pipeline/logical_plan.h
index 7d3817b7c0..9d5ba7b8ed 100644
--- a/src/trace_processor/perfetto_sql/pipeline/logical_plan.h
+++ b/src/trace_processor/perfetto_sql/pipeline/logical_plan.h
@@ -112,11 +112,33 @@ struct IntervalIntersect {
   ColumnId dur = 0;
 };
 
+// `|> INTERVAL FLATTEN [PER cols] AGGREGATE agg AS name, ...`. Cuts the rows
+// at every start and end into disjoint segments, each collapsed into one row:
+// its bounds, the keys, then one column per aggregate.
+struct IntervalFlatten {
+  enum class Function : uint8_t { kCount, kSum };
+  struct Aggregate {
+    Function function = Function::kCount;
+    // Unused by COUNT(*).
+    ColumnId column = 0;
+    ColumnId output = 0;
+  };
+  ColumnId ts = 0;
+  ColumnId dur = 0;
+  std::vector<ColumnId> keys;
+  std::vector<Aggregate> aggregates;
+  ColumnId out_ts = 0;
+  ColumnId out_dur = 0;
+};
+
 }  // namespace op
 
 // The operator a plan node holds. Passes switch on `op.index()` with one case
 // per operator, using base::variant_index<Op, T>().
-using Op = std::variant<op::Scan, op::TreeAccumulate, op::IntervalIntersect>;
+using Op = std::variant<op::Scan,
+                        op::TreeAccumulate,
+                        op::IntervalIntersect,
+                        op::IntervalFlatten>;
 
 // Stable within a plan.
 using PlanNodeId = uint32_t;
diff --git a/src/trace_processor/perfetto_sql/pipeline/logical_plan_test_utils.cc b/src/trace_processor/perfetto_sql/pipeline/logical_plan_test_utils.cc
index f250850018..276510eea8 100644
--- a/src/trace_processor/perfetto_sql/pipeline/logical_plan_test_utils.cc
+++ b/src/trace_processor/perfetto_sql/pipeline/logical_plan_test_utils.cc
@@ -100,6 +100,28 @@ std::string TreeAccumulateString(const LogicalPlan& plan,
   return out + ")";
 }
 
+std::string IntervalFlattenString(const LogicalPlan& plan,
+                                  const op::IntervalFlatten& flatten) {
+  std::string out = "IntervalFlatten(ts=#" + std::to_string(flatten.ts) +
+                    ", dur=#" + std::to_string(flatten.dur);
+  for (ColumnId key : flatten.keys) {
+    out += ", key=#" + std::to_string(key);
+  }
+  for (const auto& agg : flatten.aggregates) {
+    switch (agg.function) {
+      case op::IntervalFlatten::Function::kCount:
+        out += ", COUNT(*)";
+        break;
+      case op::IntervalFlatten::Function::kSum:
+        out += ", SUM(#" + std::to_string(agg.column) + ")";
+        break;
+    }
+    out += " -> " + ColumnString(plan, agg.output);
+  }
+  return out + ") -> " + ColumnString(plan, flatten.out_ts) + ", " +
+         ColumnString(plan, flatten.out_dur);
+}
+
 std::string IntervalIntersectString(const LogicalPlan& plan,
                                     const PlanNode& node) {
   const auto& ii = node.Cast<op::IntervalIntersect>();
@@ -135,6 +157,10 @@ std::string SubtreeString(const LogicalPlan& plan, PlanNodeId id) {
              TreeAccumulateString(plan, node.Cast<op::TreeAccumulate>()) + "\n";
     case base::variant_index<Op, op::IntervalIntersect>():
       return IntervalIntersectString(plan, node) + "\n";
+    case base::variant_index<Op, op::IntervalFlatten>():
+      return SubtreeString(plan, node.children[0]) +
+             IntervalFlattenString(plan, node.Cast<op::IntervalFlatten>()) +
+             "\n";
     default:
       PERFETTO_FATAL("Unknown operator");
   }
diff --git a/src/trace_processor/perfetto_sql/pipeline/physical_plan.cc b/src/trace_processor/perfetto_sql/pipeline/physical_plan.cc
index 7ceeefc8d5..27196e6696 100644
--- a/src/trace_processor/perfetto_sql/pipeline/physical_plan.cc
+++ b/src/trace_processor/perfetto_sql/pipeline/physical_plan.cc
@@ -16,6 +16,7 @@
 
 #include "src/trace_processor/perfetto_sql/pipeline/physical_plan.h"
 
+#include <algorithm>
 #include <cstdint>
 #include <limits>
 #include <memory>
@@ -27,13 +28,17 @@
 
 #include "perfetto/base/logging.h"
 #include "perfetto/ext/base/variant.h"
+#include "src/trace_processor/core/common/sort_types.h"
 #include "src/trace_processor/core/common/storage_types.h"
 #include "src/trace_processor/core/dataframe/types.h"
 #include "src/trace_processor/core/exec/assert_type.h"
 #include "src/trace_processor/core/exec/dataframe_scan.h"
+#include "src/trace_processor/core/exec/group_by.h"
+#include "src/trace_processor/core/exec/interval_flatten.h"
 #include "src/trace_processor/core/exec/interval_intersect.h"
 #include "src/trace_processor/core/exec/operator.h"
 #include "src/trace_processor/core/exec/pipeline.h"
+#include "src/trace_processor/core/exec/sort.h"
 #include "src/trace_processor/core/exec/tree_accumulate.h"
 #include "src/trace_processor/core/exec/tree_number_nodes.h"
 #include "src/trace_processor/core/exec/tree_order.h"
@@ -61,9 +66,13 @@ class Lowering {
 
   std::unique_ptr<ex::Source> MakeSource(const op::Scan&) const;
   void LowerTreeAccumulate(const op::TreeAccumulate&);
+  void LowerIntervalFlatten(const op::IntervalFlatten&);
 
   // Establishes the physical layout and ordering needed by a tree fold.
   void PrepareTree(const op::TreeAccumulate&);
+  // Groups rows by `keys`, ordered by `ascending` within each group, adding
+  // only the Sort and GroupBy not already established.
+  void PrepareGroups(const std::vector<ColumnId>& keys, ColumnId ascending);
   // Appends to `into` an AssertType on the column at `position` if `column`
   // is not already flat Int64. The position is the column's in the pipeline
   // `into` belongs to, which is not this one for an intersection's operand.
@@ -98,6 +107,15 @@ class Lowering {
   };
   std::optional<TreeColumns> tree_columns_;
   std::optional<op::TreeDirection> tree_order_;
+
+  // The known order of the rows: grouped by `grouped_by` (numbered in
+  // `group_column`), with `ascending` ascending within each group.
+  struct Order {
+    std::vector<ColumnId> grouped_by;
+    uint32_t group_column = 0;
+    std::vector<ColumnId> ascending;
+  };
+  Order order_;
 };
 
 void Lowering::LowerNode(PlanNodeId id) {
@@ -115,6 +133,10 @@ void Lowering::LowerNode(PlanNodeId id) {
       // its children itself.
       LowerIntervalIntersect(node.Cast<op::IntervalIntersect>(), node.children);
       return;
+    case base::variant_index<Op, op::IntervalFlatten>():
+      LowerNode(node.children[0]);
+      LowerIntervalFlatten(node.Cast<op::IntervalFlatten>());
+      return;
     default:
       PERFETTO_FATAL("Unknown operator");
   }
@@ -141,6 +163,13 @@ void Lowering::LowerScan(const op::Scan& scan) {
     Define(column.id);
   }
   out_->input_ = MakeSource(scan);
+  // A dataframe is read in order, so a column it keeps sorted ascends.
+  const auto& dataframe = base::unchecked_get<op::Scan::Dataframe>(scan.source);
+  for (uint32_t i = 0; i < scan.columns.size(); ++i) {
+    if (!dataframe.columns[i]->sort_state.Is<core::Unsorted>()) {
+      order_.ascending.push_back(scan.columns[i].id);
+    }
+  }
 }
 
 void Lowering::LowerIntervalIntersect(const op::IntervalIntersect& isect,
@@ -197,6 +226,82 @@ void Lowering::LowerIntervalIntersect(const op::IntervalIntersect& isect,
   out_->input_ = std::make_unique<ex::IntervalIntersect>(std::move(operands));
 }
 
+void Lowering::PrepareGroups(const std::vector<ColumnId>& keys,
+                             ColumnId ascending) {
+  auto has = [](const std::vector<ColumnId>& ids, ColumnId id) {
+    return std::find(ids.begin(), ids.end(), id) != ids.end();
+  };
+  bool grouped = order_.grouped_by.size() == keys.size() &&
+                 std::all_of(keys.begin(), keys.end(), [&](ColumnId key) {
+                   return has(order_.grouped_by, key);
+                 });
+  if (grouped && has(order_.ascending, ascending)) {
+    return;
+  }
+  // GroupBy is stable, so sorting first orders every group.
+  if (!order_.grouped_by.empty() || !has(order_.ascending, ascending)) {
+    ex::SortSpec sort;
+    sort.keys.push_back({Position(ascending), false});
+    operators_.push_back(std::make_unique<ex::Sort>(std::move(sort)));
+    order_ = Order();
+    order_.ascending.push_back(ascending);
+  }
+  if (!keys.empty()) {
+    std::vector<uint32_t> positions;
+    for (ColumnId key : keys) {
+      positions.push_back(Position(key));
+    }
+    operators_.push_back(std::make_unique<ex::GroupBy>(std::move(positions)));
+    order_.grouped_by = keys;
+    order_.group_column = column_count_++;
+  }
+}
+
+void Lowering::LowerIntervalFlatten(const op::IntervalFlatten& flatten) {
+  RequireInt64(flatten.ts, Position(flatten.ts), operators_);
+  RequireInt64(flatten.dur, Position(flatten.dur), operators_);
+  PrepareGroups(flatten.keys, flatten.ts);
+  ex::IntervalFlattenSpec spec;
+  spec.ts_column = Position(flatten.ts);
+  spec.dur_column = Position(flatten.dur);
+  for (ColumnId key : flatten.keys) {
+    spec.key_columns.push_back(Position(key));
+  }
+  spec.group_column = order_.group_column;
+  for (const op::IntervalFlatten::Aggregate& agg : flatten.aggregates) {
+    ex::IntervalFlattenSpec::Aggregate lowered;
+    switch (agg.function) {
+      case op::IntervalFlatten::Function::kCount:
+        lowered.function = ex::IntervalFlattenSpec::Function::kCount;
+        break;
+      case op::IntervalFlatten::Function::kSum:
+        RequireInt64(agg.column, Position(agg.column), operators_);
+        lowered.function = ex::IntervalFlattenSpec::Function::kSum;
+        lowered.column = Position(agg.column);
+        break;
+    }
+    spec.aggregates.push_back(lowered);
+  }
+  operators_.push_back(std::make_unique<ex::IntervalFlatten>(std::move(spec)));
+
+  // Its rows are the segments, laid out afresh.
+  column_count_ = 0;
+  Define(flatten.out_ts);
+  Define(flatten.out_dur);
+  for (ColumnId key : flatten.keys) {
+    Define(key);
+  }
+  for (const op::IntervalFlatten::Aggregate& agg : flatten.aggregates) {
+    Define(agg.output);
+  }
+  order_ = Order();
+  order_.grouped_by = flatten.keys;
+  order_.group_column = column_count_++;
+  order_.ascending.push_back(flatten.out_ts);
+  tree_columns_.reset();
+  tree_order_.reset();
+}
+
 void Lowering::RequireInt64(ColumnId column,
                             uint32_t position,
                             std::vector<std::unique_ptr<ex::Operator>>& into) {
@@ -225,6 +330,7 @@ void Lowering::PrepareTree(const op::TreeAccumulate& acc) {
   if (tree_order_ == acc.direction) {
     return;
   }
+  order_ = Order();
   if (acc.direction == op::TreeDirection::kUp) {
     operators_.push_back(std::make_unique<ex::TreeChildFirst>(
         tree_columns_->node, tree_columns_->parent));
diff --git a/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc b/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc
index a92d60ed8d..a4fc187c9d 100644
--- a/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc
+++ b/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc
@@ -169,10 +169,11 @@ class PlanWriter {
 
   std::string Write() {
     // Pruning can leave nodes the root does not reach.
-    std::vector<const op::TreeAccumulate*> folds;
+    std::vector<const PlanNode*> stages;
     PlanNodeId id = plan_.root;
-    while (plan_.nodes[id].Is<op::TreeAccumulate>()) {
-      folds.push_back(&plan_.nodes[id].Cast<op::TreeAccumulate>());
+    while (plan_.nodes[id].Is<op::TreeAccumulate>() ||
+           plan_.nodes[id].Is<op::IntervalFlatten>()) {
+      stages.push_back(&plan_.nodes[id]);
       id = plan_.nodes[id].children[0];
     }
     const PlanNode& source = plan_.nodes[id];
@@ -180,9 +181,20 @@ class PlanWriter {
     Available available = source.Is<op::Scan>()
                               ? WriteScan(source.Cast<op::Scan>())
                               : WriteIntervalIntersect(source);
-    w_.Size(folds.size());
-    for (auto it = folds.rbegin(); it != folds.rend(); ++it) {
-      WriteTreeAccumulate(**it, available);
+    w_.Size(stages.size());
+    for (auto it = stages.rbegin(); it != stages.rend(); ++it) {
+      const PlanNode& stage = **it;
+      w_.U8(static_cast<uint8_t>(stage.op.index()));
+      switch (stage.op.index()) {
+        case base::variant_index<Op, op::TreeAccumulate>():
+          WriteTreeAccumulate(stage.Cast<op::TreeAccumulate>(), available);
+          break;
+        case base::variant_index<Op, op::IntervalFlatten>():
+          WriteIntervalFlatten(stage.Cast<op::IntervalFlatten>(), available);
+          break;
+        default:
+          PERFETTO_FATAL("Unknown stage");
+      }
     }
     w_.Size(plan_.output.size());
     for (const NamedColumn& column : plan_.output) {
@@ -231,6 +243,32 @@ class PlanWriter {
     available.insert(available.end(), outputs.begin(), outputs.end());
   }
 
+  // Replaces `available` with the segment's columns.
+  void WriteIntervalFlatten(const op::IntervalFlatten& flatten,
+                            Available& available) {
+    w_.Position(available, flatten.ts);
+    w_.Position(available, flatten.dur);
+    w_.Size(flatten.keys.size());
+    for (ColumnId key : flatten.keys) {
+      w_.Position(available, key);
+    }
+    w_.Size(flatten.aggregates.size());
+    for (const op::IntervalFlatten::Aggregate& agg : flatten.aggregates) {
+      w_.U8(static_cast<uint8_t>(agg.function));
+      if (agg.function == op::IntervalFlatten::Function::kSum) {
+        w_.Position(available, agg.column);
+      }
+      w_.Str(plan_.columns[agg.output].name);
+    }
+    w_.Str(plan_.columns[flatten.out_ts].name);
+    w_.Str(plan_.columns[flatten.out_dur].name);
+    available = {flatten.out_ts, flatten.out_dur};
+    available.insert(available.end(), flatten.keys.begin(), flatten.keys.end());
+    for (const op::IntervalFlatten::Aggregate& agg : flatten.aggregates) {
+      available.push_back(agg.output);
+    }
+  }
+
   Available WriteIntervalIntersect(const PlanNode& node) {
     const auto& isect = node.Cast<op::IntervalIntersect>();
     w_.Str(plan_.columns[isect.ts].name);
@@ -280,9 +318,19 @@ class PlanReader {
         r_.Fail();
         return {};
     }
-    uint32_t folds = r_.Count();
-    for (uint32_t i = 0; i < folds && r_.ok(); ++i) {
-      plan_.AddNode(ReadTreeAccumulate(available), {plan_.root});
+    uint32_t stages = r_.Count();
+    for (uint32_t i = 0; i < stages && r_.ok(); ++i) {
+      switch (r_.U8()) {
+        case base::variant_index<Op, op::TreeAccumulate>():
+          plan_.AddNode(ReadTreeAccumulate(available), {plan_.root});
+          break;
+        case base::variant_index<Op, op::IntervalFlatten>():
+          plan_.AddNode(ReadIntervalFlatten(available), {plan_.root});
+          break;
+        default:
+          r_.Fail();
+          break;
+      }
     }
     // The table function declares only so many output columns.
     uint32_t outputs = r_.Count();
@@ -324,6 +372,40 @@ class PlanReader {
     return available;
   }
 
+  op::IntervalFlatten ReadIntervalFlatten(Available& available) {
+    op::IntervalFlatten flatten;
+    flatten.ts = r_.Position(available);
+    flatten.dur = r_.Position(available);
+    flatten.keys.resize(r_.Count());
+    for (ColumnId& key : flatten.keys) {
+      key = r_.Position(available);
+    }
+    flatten.aggregates.resize(r_.Count());
+    for (op::IntervalFlatten::Aggregate& agg : flatten.aggregates) {
+      switch (r_.U8()) {
+        case static_cast<uint8_t>(op::IntervalFlatten::Function::kCount):
+          agg.function = op::IntervalFlatten::Function::kCount;
+          break;
+        case static_cast<uint8_t>(op::IntervalFlatten::Function::kSum):
+          agg.function = op::IntervalFlatten::Function::kSum;
+          agg.column = r_.Position(available);
+          break;
+        default:
+          r_.Fail();
+          break;
+      }
+      agg.output = plan_.AddColumn(r_.Str(), core::Int64{});
+    }
+    flatten.out_ts = plan_.AddColumn(r_.Str(), core::Int64{});
+    flatten.out_dur = plan_.AddColumn(r_.Str(), core::Int64{});
+    available = {flatten.out_ts, flatten.out_dur};
+    available.insert(available.end(), flatten.keys.begin(), flatten.keys.end());
+    for (const op::IntervalFlatten::Aggregate& agg : flatten.aggregates) {
+      available.push_back(agg.output);
+    }
+    return flatten;
+  }
+
   op::TreeAccumulate ReadTreeAccumulate(Available& available) {
     op::TreeAccumulate acc;
     acc.direction = static_cast<op::TreeDirection>(r_.U8());
diff --git a/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc b/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc
index cb747da3d7..0e86ed0d30 100644
--- a/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc
+++ b/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc
@@ -115,6 +115,8 @@ const char* const kPipelines[] = {
     "FROM df |> TREE ACCUMULATE DOWN SUM(self) AS a, SUM(id) AS b",
     "FROM (SELECT ts, dur, cpu FROM spans WHERE dur > 1)",
     "INTERVAL INTERSECTION OF (spans AS a, spans AS b) PER cpu",
+    "FROM spans |> INTERVAL FLATTEN PER cpu AGGREGATE SUM(dur) AS d",
+    "FROM spans |> INTERVAL FLATTEN AGGREGATE COUNT(*) AS n |> SELECT ts, n",
     // Pruning removes the fold, whose node is left behind unread.
     "FROM df |> TREE ACCUMULATE UP SUM(self) AS total |> SELECT id",
 };
diff --git a/test/trace_processor/diff_tests/syntax/pipeline_tests.py b/test/trace_processor/diff_tests/syntax/pipeline_tests.py
index d09c498753..b936ce1398 100644
--- a/test/trace_processor/diff_tests/syntax/pipeline_tests.py
+++ b/test/trace_processor/diff_tests/syntax/pipeline_tests.py
@@ -96,6 +96,118 @@ def test_tree_accumulate_matches_closure(self):
         74228,0
         """))
 
+  def test_interval_flatten(self):
+    return DiffTestBlueprint(
+        trace=TextProto(r''),
+        query="""
+        PERFETTO PRAGMA pipelines = 1;
+        CREATE PERFETTO TABLE spans AS
+        SELECT 10 AS ts, 100 AS dur, 1 AS cpu, 5 AS weight
+        UNION ALL SELECT 20, 40, 1, 1
+        UNION ALL SELECT 30, 120, 1, NULL
+        UNION ALL SELECT 60, 0, 1, 2
+        UNION ALL SELECT 200, 10, 2, 3
+        UNION ALL SELECT 210, 10, 2, 4;
+
+        CREATE PERFETTO TABLE flat AS
+        FROM spans
+        |> INTERVAL FLATTEN PER cpu
+           AGGREGATE COUNT(*) AS n, SUM(weight) AS weight;
+
+        SELECT * FROM flat ORDER BY cpu, ts, dur;
+        """,
+        out=Csv("""
+        "ts","dur","cpu","n","weight"
+        10,10,1,1,5
+        20,10,1,2,6
+        30,30,1,3,6
+        60,0,1,3,7
+        60,50,1,2,5
+        110,40,1,1,"[NULL]"
+        200,10,2,1,3
+        210,10,2,1,4
+        """))
+
+  # Must match interval_self_intersect, which this is meant to replace.
+  def test_interval_flatten_matches_self_intersect(self):
+    return DiffTestBlueprint(
+        trace=DataPath('chrome_input_with_frame_view.pftrace'),
+        query="""
+        PERFETTO PRAGMA pipelines = 1;
+        INCLUDE PERFETTO MODULE intervals.intersect;
+
+        CREATE PERFETTO TABLE piped AS
+        FROM (SELECT ts, dur FROM slice WHERE dur > 0)
+        |> INTERVAL FLATTEN AGGREGATE COUNT(*) AS n;
+
+        CREATE PERFETTO TABLE macro AS
+        SELECT ts, dur, sum(NOT interval_ends_at_ts) AS n
+        FROM interval_self_intersect!(
+          (SELECT id, ts, dur FROM slice WHERE dur > 0)
+        )
+        GROUP BY group_id
+        HAVING n > 0;
+
+        SELECT
+          (SELECT count(*) FROM piped) AS rows,
+          (
+            SELECT count(*) FROM (
+              SELECT ts, dur, n FROM piped
+              EXCEPT SELECT ts, dur, n FROM macro
+            )
+          ) + (
+            SELECT count(*) FROM (
+              SELECT ts, dur, n FROM macro
+              EXCEPT SELECT ts, dur, n FROM piped
+            )
+          ) AS mismatches;
+        """,
+        out=Csv("""
+        "rows","mismatches"
+        107295,0
+        """))
+
+  # Per key, sum(n * dur) over segments equals sum(dur) over rows.
+  def test_interval_flatten_keys_over_many_batches(self):
+    return DiffTestBlueprint(
+        trace=DataPath('chrome_input_with_frame_view.pftrace'),
+        query="""
+        PERFETTO PRAGMA pipelines = 1;
+
+        CREATE PERFETTO TABLE piped AS
+        FROM (
+          SELECT ts, dur, track_id, IIF(depth % 2 = 0, 'even', 'odd') AS parity
+          FROM slice
+          WHERE dur > 0
+        )
+        |> INTERVAL FLATTEN PER track_id, parity AGGREGATE COUNT(*) AS n;
+
+        CREATE PERFETTO TABLE expected AS
+        SELECT
+          track_id,
+          IIF(depth % 2 = 0, 'even', 'odd') AS parity,
+          sum(dur) AS covered
+        FROM slice
+        WHERE dur > 0
+        GROUP BY track_id, parity;
+
+        SELECT
+          (SELECT count(*) FROM expected) AS keys,
+          (
+            SELECT count(*) FROM expected e
+            FULL JOIN (
+              SELECT track_id, parity, sum(n * dur) AS covered
+              FROM piped
+              GROUP BY track_id, parity
+            ) p USING (track_id, parity)
+            WHERE e.covered IS NOT p.covered
+          ) AS mismatches;
+        """,
+        out=Csv("""
+        "keys","mismatches"
+        101,0
+        """))
+
   def test_pipeline_errors_name_the_problem(self):
     return DiffTestBlueprint(
         trace=TextProto(r''),