tp: have the executor end every step's run when its stream ends

**Why**

A step let go of what it held only if it remembered to clear itself,
and almost none did. GraphTree, the interval operators and the tree
transforms kept everything they built until the next run. One
`OperatorState` mixed per-call scratch, rows held for the whole input
and the step's status, so nothing said what a step may keep or for how
long.

**Changes**

- `operator.h` states the rules every step keeps to. Each field of a
  state declares its lifetime: `Held<T>` or `RunValue<T>` for the run,
  `CallScratch<T>` for one call, and anything else is set when the
  state is made.
- `Held<T>` and `RunValue<T>` register with their state. The pipeline
  ends their run as soon as the step's stream ends, and again before a
  source is restarted. `Reset()` and `ResetEachRun` are gone.
- States are made once and reused, because the same pipeline is often
  rerun for small queries. Ending a run keeps up to 64 KiB of each
  container's storage and releases anything larger.
- The pipeline merges its states' registries by segment, so ending a
  run is one pass with no call per state.
- A batch a step outputs never points into its state, so a run can end
  while its batches are still read. `GroupedSort` copies its group
  numbers into a buffer instead of lending them.
- Each state's fields are grouped by what they do.

**Results**

`INCLUDE` of `heap_graph_dominator_tree` on
`test/data/heap_graph_object_for_benchmarks.pftrace`, release build:

| | Peak RSS | Buffers |
|---|---:|---:|
| #8331 | 305 MB | 57 MB |
| This PR | 270 MB | 53 MB |

Every heap graph table is byte-identical. Rerunning a one-row pipeline
costs about 2-3 ns more per step that holds state. Large runs are
unchanged.

**Testing**

- New `StateTest.AStepsRunEndsWithItsStream`.
- The affected unit test suites pass.

Author: LalitMaganti

BUILD

@@ -2688,6 +2688,7 @@ perfetto_filegroup(
         "src/trace_processor/core/exec/row_cursor.h",
         "src/trace_processor/core/exec/row_store.cc",
         "src/trace_processor/core/exec/row_store.h",
+        "src/trace_processor/core/exec/run_held.h",
         "src/trace_processor/core/exec/selection.cc",
         "src/trace_processor/core/exec/selection.h",
         "src/trace_processor/core/exec/sort.cc",

src/trace_processor/core/exec/BUILD.gn

@@ -78,6 +78,7 @@ source_set("exec") {
     "row_cursor.h",
     "row_store.cc",
     "row_store.h",
+    "run_held.h",
     "selection.cc",
     "selection.h",
     "sort.cc",

src/trace_processor/core/exec/aggregate_function.cc

@@ -25,6 +25,7 @@
 #include "perfetto/ext/base/utils.h"
 #include "src/trace_processor/core/common/storage_types.h"
 #include "src/trace_processor/core/exec/column_chunk.h"
+#include "src/trace_processor/core/exec/run_held.h"
 
 namespace perfetto::trace_processor::core::exec {
 
@@ -90,6 +91,11 @@ uint8_t* SeenBytes::For(uint32_t groups, bool nulls) {
   return bytes_.data();
 }
 
+void SeenBytes::EndRun() {
+  Clear();
+  EndRunOf(&bytes_);
+}
+
 void SeenBytes::Clear() {
   bytes_.clear();
   tracking_ = false;

src/trace_processor/core/exec/aggregate_function.h

@@ -24,6 +24,7 @@
 #include "src/trace_processor/core/exec/column_view.h"
 #include "src/trace_processor/core/exec/context.h"
 #include "src/trace_processor/core/exec/operator.h"
+#include "src/trace_processor/core/exec/run_held.h"
 #include "src/trace_processor/core/exec/selection.h"
 #include "src/trace_processor/core/util/bit_vector.h"
 #include "src/trace_processor/core/util/flex_vector.h"
@@ -71,6 +72,12 @@ class AggregateInputLoader {
             uint32_t count,
             AggregateInput* out);
 
+  // Ends a run (see RunHeld).
+  void EndRun() {
+    EndRunOf(&values_);
+    EndRunOf(&valid_);
+  }
+
  private:
   FlexVector<int64_t> values_;
   FlexVector<uint64_t> valid_;
@@ -96,6 +103,8 @@ class SeenBytes {
   // groups in all, whose input holds nulls if `nulls`.
   uint8_t* For(uint32_t groups, bool nulls);
   void Clear();
+  // Clear(), keeping only small allocations for the next run (see RunHeld).
+  void EndRun();
 
  private:
   FlexVector<uint8_t> bytes_;

src/trace_processor/core/exec/assert_type.cc

@@ -175,10 +175,6 @@ base::Status AssertType::status(const OperatorState& state) const {
   return state.Cast<const State>().status;
 }
 
-void AssertType::State::Reset() {
-  status = base::OkStatus();
-}
-
 bool AssertType::Widen(const ColumnView& column,
                        const Selection& selection,
                        ColumnChunk& chunk,

src/trace_processor/core/exec/assert_type.h

@@ -29,6 +29,7 @@
 #include "src/trace_processor/core/exec/context.h"
 #include "src/trace_processor/core/exec/operator.h"
 #include "src/trace_processor/core/exec/row_batch.h"
+#include "src/trace_processor/core/exec/run_held.h"
 
 namespace perfetto::trace_processor::core::exec {
 
@@ -61,12 +62,11 @@ class AssertType final : public Transform {
   friend class PipelineFormatter;
 
   struct State : OperatorState {
-    explicit State(Context& c) : OperatorState(ResetEachRun{}), context(&c) {}
+    explicit State(Context& c) : context(&c) {}
     ~State() override;
-    void Reset() override;
     // What the converted column is written in.
     Context* context;
-    base::Status status = base::OkStatus();
+    Held<base::Status> status{this};
   };
 
   bool Widen(const ColumnView&, const Selection&, ColumnChunk&, State&) const;

src/trace_processor/core/exec/batch_buffer.h

@@ -44,6 +44,8 @@ class BatchBuffer {
       column.nullable = false;
     }
   }
+  // Ends a run (see RunHeld): the buffers go back to the pool.
+  void EndRun() { Clear(); }
 
  private:
   struct Column {

src/trace_processor/core/exec/breaker.cc

@@ -60,11 +60,6 @@ OpResult Breaker::Finish(RowBatch& out, OperatorState& state) const {
   return s.status.ok() ? OpResult::kNeedMoreInput : OpResult::kError;
 }
 
-void Breaker::State::Reset() {
-  status = base::OkStatus();
-  filled = false;
-}
-
 base::Status Breaker::status(const OperatorState& state) const {
   const State& s = state.Cast<const State>();
   return s.status;

src/trace_processor/core/exec/breaker.h

@@ -22,6 +22,7 @@
 #include "perfetto/base/status.h"
 #include "src/trace_processor/core/exec/operator.h"
 #include "src/trace_processor/core/exec/row_batch.h"
+#include "src/trace_processor/core/exec/run_held.h"
 
 namespace perfetto::trace_processor::core::exec {
 
@@ -31,14 +32,15 @@ namespace perfetto::trace_processor::core::exec {
 class Breaker : public Operator {
  public:
   struct State : OperatorState {
-    State() : OperatorState(ResetEachRun{}) {}
+    State() = default;
     ~State() override;
-    void Reset() override;
 
-    base::Status status = base::OkStatus();
-    bool filled = false;
     // What the columns served are filled in.
     Context* context = nullptr;
+
+    Held<base::Status> status{this};
+    // Whether Finalize() has run.
+    RunValue<bool> filled{this, false};
   };
 
   ~Breaker() override;

src/trace_processor/core/exec/breaker_benchmark.cc

@@ -70,7 +70,7 @@ void Run(benchmark::State& state,
     while (op.Finish(out, *op_state) == OpResult::kHaveMoreOutput) {
       benchmark::DoNotOptimize(out.size());
     }
-    op_state->Reset();
+    op_state->EndRun();
   }
   state.SetItemsProcessed(static_cast<int64_t>(state.iterations()) * rows);
 }