tp: add Sort and GroupBy operators to the executor

Android.bp

@@ -17822,13 +17822,15 @@ filegroup {
         "src/trace_processor/core/exec/column_chunk.cc",
         "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_intersect.cc",
         "src/trace_processor/core/exec/key_encoder.cc",
         "src/trace_processor/core/exec/operator.cc",
         "src/trace_processor/core/exec/pipeline.cc",
         "src/trace_processor/core/exec/row_batch.cc",
         "src/trace_processor/core/exec/row_cursor.cc",
         "src/trace_processor/core/exec/row_store.cc",
+        "src/trace_processor/core/exec/sort.cc",
         "src/trace_processor/core/exec/tree_accumulate.cc",
         "src/trace_processor/core/exec/tree_number_nodes.cc",
         "src/trace_processor/core/exec/tree_order.cc",
@@ -17848,11 +17850,13 @@ filegroup {
         "src/trace_processor/core/exec/breaker_unittest.cc",
         "src/trace_processor/core/exec/dataframe_scan_unittest.cc",
         "src/trace_processor/core/exec/executor_contract_unittest.cc",
+        "src/trace_processor/core/exec/group_by_unittest.cc",
         "src/trace_processor/core/exec/interval_intersect_unittest.cc",
         "src/trace_processor/core/exec/key_encoder_unittest.cc",
         "src/trace_processor/core/exec/operator_unittest.cc",
         "src/trace_processor/core/exec/row_batch_unittest.cc",
         "src/trace_processor/core/exec/row_store_unittest.cc",
+        "src/trace_processor/core/exec/sort_unittest.cc",
         "src/trace_processor/core/exec/tree_accumulate_unittest.cc",
         "src/trace_processor/core/exec/tree_number_nodes_unittest.cc",
         "src/trace_processor/core/exec/tree_order_unittest.cc",

BUILD

@@ -2585,11 +2585,14 @@ perfetto_filegroup(
         "src/trace_processor/core/exec/column_view.h",
         "src/trace_processor/core/exec/dataframe_scan.cc",
         "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_columns.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",
         "src/trace_processor/core/exec/key_encoder.h",
+        "src/trace_processor/core/exec/layout_column.h",
         "src/trace_processor/core/exec/operator.cc",
         "src/trace_processor/core/exec/operator.h",
         "src/trace_processor/core/exec/pipeline.cc",
@@ -2601,6 +2604,8 @@ perfetto_filegroup(
         "src/trace_processor/core/exec/row_selection.h",
         "src/trace_processor/core/exec/row_store.cc",
         "src/trace_processor/core/exec/row_store.h",
+        "src/trace_processor/core/exec/sort.cc",
+        "src/trace_processor/core/exec/sort.h",
         "src/trace_processor/core/exec/tree_accumulate.cc",
         "src/trace_processor/core/exec/tree_accumulate.h",
         "src/trace_processor/core/exec/tree_number_nodes.cc",

src/trace_processor/core/exec/BUILD.gn

@@ -30,11 +30,14 @@ source_set("exec") {
     "column_view.h",
     "dataframe_scan.cc",
     "dataframe_scan.h",
+    "group_by.cc",
+    "group_by.h",
     "interval_columns.h",
     "interval_intersect.cc",
     "interval_intersect.h",
     "key_encoder.cc",
     "key_encoder.h",
+    "layout_column.h",
     "operator.cc",
     "operator.h",
     "pipeline.cc",
@@ -46,6 +49,8 @@ source_set("exec") {
     "row_selection.h",
     "row_store.cc",
     "row_store.h",
+    "sort.cc",
+    "sort.h",
     "tree_accumulate.cc",
     "tree_accumulate.h",
     "tree_number_nodes.cc",
@@ -82,11 +87,13 @@ perfetto_unittest_source_set("unittests") {
     "breaker_unittest.cc",
     "dataframe_scan_unittest.cc",
     "executor_contract_unittest.cc",
+    "group_by_unittest.cc",
     "interval_intersect_unittest.cc",
     "key_encoder_unittest.cc",
     "operator_unittest.cc",
     "row_batch_unittest.cc",
     "row_store_unittest.cc",
+    "sort_unittest.cc",
     "tree_accumulate_unittest.cc",
     "tree_number_nodes_unittest.cc",
     "tree_order_unittest.cc",

src/trace_processor/core/exec/group_by.cc

@@ -0,0 +1,120 @@
+/*
+ * 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/group_by.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "perfetto/base/status.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/span.h"
+
+namespace perfetto::trace_processor::core::exec {
+
+GroupBy::GroupBy(std::vector<uint32_t> key_columns)
+    : key_columns_(std::move(key_columns)) {}
+GroupBy::~GroupBy() = default;
+GroupBy::State::~State() = default;
+
+std::unique_ptr<Breaker::State> GroupBy::CreateState() const {
+  return std::make_unique<State>();
+}
+
+bool GroupBy::Consume(const RowBatch& in, Breaker::State& state) const {
+  auto& s = static_cast<State&>(state);
+  uint32_t count = in.size();
+  if (count == 0) {
+    return true;
+  }
+  if (std::optional<uint32_t> bad = s.keys.Encode(in, key_columns_)) {
+    s.status = base::ErrStatus(
+        "GROUP BY: key %u must be a column holding one type", *bad + 1);
+    return false;
+  }
+  for (uint32_t row = 0; row < count; ++row) {
+    std::string_view key = s.keys.Key(row);
+    uint32_t group;
+    if (const uint32_t* found = s.group_of.Find(key)) {
+      group = *found;
+    } else {
+      group = static_cast<uint32_t>(s.group_of.size());
+      s.group_of.Insert(std::string(key), group);
+    }
+    s.groups.push_back(group);
+  }
+  s.status = s.rows.Append(in);
+  return s.status.ok();
+}
+
+bool GroupBy::Finalize(Breaker::State& state) const {
+  auto& s = static_cast<State&>(state);
+  // A counting sort on the group.
+  auto groups = static_cast<uint32_t>(s.group_of.size());
+  auto next = FlexVector<uint32_t>::CreateFilled(groups, 0);
+  for (uint32_t group : s.groups) {
+    ++next[group];
+  }
+  uint32_t start = 0;
+  for (uint32_t& at : next) {
+    uint32_t rows = at;
+    at = start;
+    start += rows;
+  }
+  s.order.resize(s.groups.size());
+  s.ordered_groups.resize(s.groups.size());
+  for (uint32_t row = 0; row < s.groups.size(); ++row) {
+    uint32_t at = next[s.groups[row]]++;
+    s.order[at] = row;
+    s.ordered_groups[at] = s.groups[row];
+  }
+  return true;
+}
+
+bool GroupBy::Serve(RowBatch& out, Breaker::State& state) const {
+  auto& s = static_cast<State&>(state);
+  if (s.emitted == s.rows.size()) {
+    return false;
+  }
+  uint32_t count = std::min(kMaxBatchRows, s.rows.size() - s.emitted);
+  const uint32_t* begin = s.order.data() + s.emitted;
+  count = s.rows.View(&out, Span<const uint32_t>(begin, begin + count));
+  ColumnView group =
+      ColumnView::Reference(StorageType{Uint32{}}, s.ordered_groups.data());
+  group.SetRange(s.emitted);
+  out.AddColumn(group);
+  s.emitted += count;
+  return true;
+}
+
+void GroupBy::Reset(Breaker::State& state) const {
+  auto& s = static_cast<State&>(state);
+  s.group_of.Clear();
+  s.groups.clear();
+  s.rows.Clear();
+  s.order.clear();
+  s.ordered_groups.clear();
+  s.emitted = 0;
+}
+
+}  // namespace perfetto::trace_processor::core::exec

src/trace_processor/core/exec/group_by.h

@@ -0,0 +1,71 @@
+/*
+ * 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_GROUP_BY_H_
+#define SRC_TRACE_PROCESSOR_CORE_EXEC_GROUP_BY_H_
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "perfetto/ext/base/flat_hash_map.h"
+#include "src/trace_processor/core/exec/breaker.h"
+#include "src/trace_processor/core/exec/key_encoder.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/flex_vector.h"
+
+namespace perfetto::trace_processor::core::exec {
+
+// Puts rows sharing their keys next to each other. Stable, so an order the
+// input had holds within each group. Appends each row's group number as a
+// Uint32 column, so what follows finds group boundaries without comparing
+// keys.
+class GroupBy : public Breaker {
+ public:
+  explicit GroupBy(std::vector<uint32_t> key_columns);
+  ~GroupBy() override;
+  BatchPreference batch_preference() const override {
+    return BatchPreference::kThroughput;
+  }
+
+ private:
+  struct State : Breaker::State {
+    ~State() override;
+
+    KeyEncoder keys;
+    base::FlatHashMapV2<std::string, uint32_t> group_of;
+    FlexVector<uint32_t> groups;
+    RowStore rows;
+    FlexVector<uint32_t> order;
+    FlexVector<uint32_t> ordered_groups;
+    uint32_t emitted = 0;
+  };
+
+  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;
+
+  std::vector<uint32_t> key_columns_;
+};
+
+}  // namespace perfetto::trace_processor::core::exec
+
+#endif  // SRC_TRACE_PROCESSOR_CORE_EXEC_GROUP_BY_H_