From 09f9c28be4427b7086cafeb75117cc3bf5104ff7 Mon Sep 17 00:00:00 2001 From: Lalit Maganti Date: Wed, 30 Sep 2026 00:52:23 +0100 Subject: [PATCH] tp: add INTERVAL FLATTEN to pipelines FROM spans |> INTERVAL FLATTEN PER cpu AGGREGATE COUNT(*) AS n, SUM(weight) AS w Cuts overlapping rows at every start and end into disjoint segments, one row per segment with its bounds, the PER keys and the aggregates. A row of no width becomes a segment of no width counting the rows spanning it. Bounds follow INTERVAL INTERSECTION: null ts or dur is skipped, negative is an error. Only COUNT(*) and SUM for now. The operator needs its input grouped by the keys and ordered by ts within each group. Lowering tracks what order is known and adds Sort and GroupBy only when needed, e.g. no Sort for a dataframe sorted on ts. Its output keeps that order, so a following stage needs neither. Its heap of interval ends uses the sifts in core/util/heap.h, shared with the trace sorter (#7711). --- Android.bp | 2 + BUILD | 3 + src/trace_processor/core/exec/BUILD.gn | 3 + .../core/exec/interval_flatten.cc | 324 ++++++++++++++++++ .../core/exec/interval_flatten.h | 134 ++++++++ .../core/exec/interval_flatten_benchmark.cc | 90 +++++ src/trace_processor/core/util/BUILD.gn | 2 + src/trace_processor/core/util/heap.h | 73 ++++ .../core/util/heap_unittest.cc | 76 ++++ .../perfetto_sql_connection_unittest.cc | 85 ++++- .../perfetto_sql/pipeline/column_pruning.cc | 29 ++ .../perfetto_sql/pipeline/compiler.cc | 77 ++++- .../perfetto_sql/pipeline/logical_plan.h | 24 +- .../pipeline/logical_plan_test_utils.cc | 26 ++ .../perfetto_sql/pipeline/physical_plan.cc | 106 ++++++ .../pipeline/plan_serialization.cc | 100 +++++- .../pipeline/plan_serialization_unittest.cc | 2 + .../diff_tests/syntax/pipeline_tests.py | 112 ++++++ 18 files changed, 1253 insertions(+), 15 deletions(-) create mode 100644 src/trace_processor/core/exec/interval_flatten.cc create mode 100644 src/trace_processor/core/exec/interval_flatten.h create mode 100644 src/trace_processor/core/exec/interval_flatten_benchmark.cc create mode 100644 src/trace_processor/core/util/heap.h create mode 100644 src/trace_processor/core/util/heap_unittest.cc diff --git a/Android.bp b/Android.bp index 7ab674c583f..1bf21a82177 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 6cab5e44009..adcc6f0480b 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 a50f3b01dad..3c0fdded52d 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 00000000000..bd113f19f05 --- /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 +#include +#include +#include +#include +#include + +#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(); +} + +constexpr auto kEndsLater = [](const auto& a, const auto& b) { + return a.end > b.end; +}; + +template +void PushEnd(FlexVector& heap, Live live) { + heap.push_back(live); + HeapSiftUp(heap.data(), heap.size() - 1, live, kEndsLater); +} + +template +void PopEnd(FlexVector& 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 IntervalFlatten::CreateState() const { + auto state = std::make_unique(); + 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); + 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 ts(in.column(spec_.ts_column)); + FlatColumnReader dur(in.column(spec_.dur_column)); + base::SmallVector, 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(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::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::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); + if (s.in_group) { + EndGroup(s); + } + return true; +} + +bool IntervalFlatten::Serve(RowBatch& out, Breaker::State& state) const { + auto& s = static_cast(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); + 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 00000000000..d1be8f35af9 --- /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 +#include +#include + +#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 key_columns; + // As GroupBy appends. Unused without keys. + uint32_t group_column = 0; + std::vector 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 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 ends; + // Sum values of live rows, by slot. + uint32_t slots = 0; + FlexVector slot_sums; + FlexVector 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 segment_ts; + FlexVector segment_dur; + FlexVector segment_groups; + FlexVector segment_key_rows; + uint32_t key_row = 0; + // What every kCount aggregate holds. + FlexVector segment_counts; + struct SegmentSums { + FlexVector values; + BitVector present; + }; + std::vector segment_sums; + uint32_t served = 0; + RowBatch served_keys; + }; + + std::unique_ptr 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 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 00000000000..2f25b8e95e6 --- /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 + +#include +#include +#include +#include + +#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 ts; + std::vector dur; + std::vector 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 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* 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 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(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 6077b814234..cd0adfdadce 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 00000000000..881acc50b51 --- /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 + +#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 +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 +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 00000000000..e86d0e1f76b --- /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 +#include +#include +#include + +#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& heap, uint32_t value) { + heap.push_back(value); + HeapSiftUp(heap.data(), heap.size() - 1, value, kLater); +} + +uint32_t Pop(std::vector& 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 heap; + std::vector 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 d09e021f9e1..b7d58cfb0a8 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 24a71844ea0..8d8f95c6dee 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(); + 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(): return PruneIntervalIntersect(plan, id, needed); + case base::variant_index(): + 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 87ba5c94611..ccbfd3d4016 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 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 CompileSum(uint32_t agg_id, uint32_t expr); base::StatusOr 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(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(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(p_, stage); + + op::IntervalFlatten flatten; + ASSIGN_OR_RETURN(flatten.ts, Resolve("ts", stage)); + ASSIGN_OR_RETURN(flatten.dur, Resolve("dur", stage)); + + std::vector> keys; + if (syntaqlite_node_is_present(n->per)) { + const auto* per = Node(p_, n->per); + std::vector 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(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> aggregates; + const auto* list = Node(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(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 7d3817b7c0d..9d5ba7b8ed3 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 keys; + std::vector 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(). -using Op = std::variant; +using Op = std::variant; // 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 f2508500185..276510eea8e 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(); @@ -135,6 +157,10 @@ std::string SubtreeString(const LogicalPlan& plan, PlanNodeId id) { TreeAccumulateString(plan, node.Cast()) + "\n"; case base::variant_index(): return IntervalIntersectString(plan, node) + "\n"; + case base::variant_index(): + return SubtreeString(plan, node.children[0]) + + IntervalFlattenString(plan, node.Cast()) + + "\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 7ceeefc8d53..27196e66966 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 #include #include #include @@ -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 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& 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 tree_columns_; std::optional 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 grouped_by; + uint32_t group_column = 0; + std::vector ascending; + }; + Order order_; }; void Lowering::LowerNode(PlanNodeId id) { @@ -115,6 +133,10 @@ void Lowering::LowerNode(PlanNodeId id) { // its children itself. LowerIntervalIntersect(node.Cast(), node.children); return; + case base::variant_index(): + LowerNode(node.children[0]); + LowerIntervalFlatten(node.Cast()); + 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(scan.source); + for (uint32_t i = 0; i < scan.columns.size(); ++i) { + if (!dataframe.columns[i]->sort_state.Is()) { + 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(std::move(operands)); } +void Lowering::PrepareGroups(const std::vector& keys, + ColumnId ascending) { + auto has = [](const std::vector& 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(std::move(sort))); + order_ = Order(); + order_.ascending.push_back(ascending); + } + if (!keys.empty()) { + std::vector positions; + for (ColumnId key : keys) { + positions.push_back(Position(key)); + } + operators_.push_back(std::make_unique(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(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>& 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( 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 a92d60ed8d6..a4fc187c9df 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 folds; + std::vector stages; PlanNodeId id = plan_.root; - while (plan_.nodes[id].Is()) { - folds.push_back(&plan_.nodes[id].Cast()); + while (plan_.nodes[id].Is() || + plan_.nodes[id].Is()) { + 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() ? WriteScan(source.Cast()) : 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(stage.op.index())); + switch (stage.op.index()) { + case base::variant_index(): + WriteTreeAccumulate(stage.Cast(), available); + break; + case base::variant_index(): + WriteIntervalFlatten(stage.Cast(), 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(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(); 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(): + plan_.AddNode(ReadTreeAccumulate(available), {plan_.root}); + break; + case base::variant_index(): + 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(op::IntervalFlatten::Function::kCount): + agg.function = op::IntervalFlatten::Function::kCount; + break; + case static_cast(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(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 cb747da3d78..0e86ed0d30c 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 d09c4987538..b936ce1398c 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''),