From dc839e8e5fdb169e6556bd5f8e1e106e3c3199b8 Mon Sep 17 00:00:00 2001 From: Lalit Maganti Date: Sun, 27 Sep 2026 06:12:05 +0100 Subject: [PATCH] tp: run pipelines from their serialized plans A pipeline statement currently runs through a TEMP virtual table created for it, with its execution plan bound to the statement as a pointer. The plan only lives as long as that one prepared statement, so SQL holding a pipeline can't be stored and run later, as views, functions and triggers do. That blocks letting pipelines appear inside ordinary SQL. Instead, write the pipeline's optimized logical plan into the SQL itself: SELECT c0 AS ts, c1 AS dur FROM __intrinsic_pipeline(X'...') `__intrinsic_pipeline` is a single built-in table function which loads the plan and lowers it when it runs. The plan is everything needed to run the pipeline, so the SQL works wherever it ends up, with nothing to keep alive or clean up alongside it. This removes the per-statement tables and the machinery that retired and dropped them. The plan is stored after optimization, so loading it is a matter of reading it back and lowering it: nothing is parsed or compiled again. Dataframes are looked up again by name, and a plan whose dataframes have changed shape since is refused. Anyone can write such bytes into SQL, so the layout can only describe plans the compiler could have produced: a source followed by the folds above it, each naming the columns it reads by their position among those available to it. Reading one back needs only bounds checks, and bytes which are not a plan are refused rather than run. Nothing changes for users yet: pipelines still appear only as whole statements or as the body of a CREATE PERFETTO TABLE. --- Android.bp | 2 + BUILD | 2 + .../engine/perfetto_sql_connection.cc | 36 +- .../engine/perfetto_sql_connection.h | 9 +- .../perfetto_sql_connection_unittest.cc | 151 ++---- .../perfetto_sql/engine/pipeline_module.cc | 259 ++++------- .../perfetto_sql/engine/pipeline_module.h | 64 +-- .../perfetto_sql/pipeline/BUILD.gn | 8 +- .../pipeline/physical_plan_unittest.cc | 34 +- .../pipeline/plan_serialization.cc | 438 ++++++++++++++++++ .../pipeline/plan_serialization.h | 45 ++ .../pipeline/plan_serialization_unittest.cc | 196 ++++++++ .../sqlite/sqlite_connection.cc | 10 +- .../sqlite/sqlite_connection.h | 15 +- 14 files changed, 917 insertions(+), 352 deletions(-) create mode 100644 src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc create mode 100644 src/trace_processor/perfetto_sql/pipeline/plan_serialization.h create mode 100644 src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc diff --git a/Android.bp b/Android.bp index 7edd8838e29..862fd6c4d21 100644 --- a/Android.bp +++ b/Android.bp @@ -19081,6 +19081,7 @@ filegroup { "src/trace_processor/perfetto_sql/pipeline/catalog.cc", "src/trace_processor/perfetto_sql/pipeline/column_pruning.cc", "src/trace_processor/perfetto_sql/pipeline/compiler.cc", + "src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc", ], } @@ -19110,6 +19111,7 @@ filegroup { name: "perfetto_src_trace_processor_perfetto_sql_pipeline_unittests", srcs: [ "src/trace_processor/perfetto_sql/pipeline/physical_plan_unittest.cc", + "src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc", ], } diff --git a/BUILD b/BUILD index a5f594afaf9..8ecb59b1b56 100644 --- a/BUILD +++ b/BUILD @@ -3881,6 +3881,8 @@ perfetto_filegroup( "src/trace_processor/perfetto_sql/pipeline/compiler.cc", "src/trace_processor/perfetto_sql/pipeline/compiler.h", "src/trace_processor/perfetto_sql/pipeline/logical_plan.h", + "src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc", + "src/trace_processor/perfetto_sql/pipeline/plan_serialization.h", ], ) diff --git a/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.cc b/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.cc index 0706771d67a..8070ed21805 100644 --- a/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.cc +++ b/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.cc @@ -57,6 +57,7 @@ #include "src/trace_processor/perfetto_sql/parser/perfetto_sql_parser.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" #include "src/trace_processor/perfetto_sql/pipeline/physical_plan.h" +#include "src/trace_processor/perfetto_sql/pipeline/plan_serialization.h" #include "src/trace_processor/sqlite/bindings/sqlite_column.h" #include "src/trace_processor/sqlite/bindings/sqlite_type.h" #include "src/trace_processor/sqlite/bindings/sqlite_value.h" @@ -359,7 +360,7 @@ PerfettoSqlConnection::PerfettoSqlConnection( { auto ctx = std::make_unique(); ctx->pool = pool_; - pipeline_context_ = ctx.get(); + ctx->connection = this; RegisterVirtualTableModule(PipelineModule::kName, std::move(ctx)); } @@ -701,9 +702,8 @@ PerfettoSqlConnection::ProcessFrame(size_t frame_idx) { std::holds_alternative(stmt))) { source_to_prepare = parser->TakeStatementSql(); } else if (std::holds_alternative(stmt)) { - auto pipeline = - std::get(parser->TakeStatement()); - pipeline_plan = std::move(pipeline.plan); + pipeline_plan = std::move( + std::get(parser->TakeStatement()).plan); source_to_prepare = parser->TakeStatementSql(); } else { is_dummy = true; @@ -714,8 +714,8 @@ PerfettoSqlConnection::ProcessFrame(size_t frame_idx) { { PERFETTO_TP_TRACE(metatrace::Category::QUERY_TIMELINE, "QUERY_PREPARE"); if (pipeline_plan) { - ASSIGN_OR_RETURN(next_stmt, PreparePipeline(std::move(*pipeline_plan), - *source_to_prepare)); + ASSIGN_OR_RETURN(next_stmt, + PreparePipeline(*pipeline_plan, *source_to_prepare)); } else { auto stmt_result = connection_->PrepareStatement(std::move(*source_to_prepare)); @@ -979,9 +979,9 @@ base::Status PerfettoSqlConnection::ExecuteCreateTable( [&create_table](metatrace::Record* record) { record->AddArg("table_name", create_table.name); }); - auto* logical = std::get_if(&create_table.body); + const auto* logical = std::get_if(&create_table.body); base::StatusOr stmt_or = - logical ? PreparePipeline(std::move(*logical), statement_sql) + logical ? PreparePipeline(*logical, statement_sql) : connection_->PrepareStatement( std::move(std::get(create_table.body))); ASSIGN_OR_RETURN(auto stmt, std::move(stmt_or)); @@ -1049,14 +1049,26 @@ base::Status PerfettoSqlConnection::ExecuteCreateTable( } base::StatusOr -PerfettoSqlConnection::PreparePipeline(pipeline::LogicalPlan logical, +PerfettoSqlConnection::PreparePipeline(const pipeline::LogicalPlan& plan, const SqlSource& source) { - PERFETTO_TP_TRACE(metatrace::Category::QUERY_TIMELINE, "PIPELINE_PLAN"); + auto sql = PipelineModule::SelectFrom(plan); + if (!sql.ok()) { + return base::ErrStatus("%s%s", source.AsTraceback(0).c_str(), + sql.status().c_message()); + } + return connection_->PrepareStatement(source.RewriteAllIgnoreExisting( + SqlSource::FromTraceProcessorImplementation(std::move(*sql)))); +} + +base::StatusOr> +PerfettoSqlConnection::LoadPipeline(std::string_view serialized) { + PERFETTO_TP_TRACE(metatrace::Category::QUERY_TIMELINE, "PIPELINE_LOAD"); + ASSIGN_OR_RETURN(pipeline::LogicalPlan plan, + pipeline::DeserializePlan(serialized, *catalog_)); pipeline::LowerEnvironment env; env.connection = connection_.get(); env.pool = pool_; - return PipelineModule::Prepare(connection_.get(), pipeline_context_, - pipeline::Lower(logical, env), source); + return pipeline::Lower(plan, env); } base::Status PerfettoSqlConnection::ExecuteCreateView( diff --git a/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.h b/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.h index 3b368fd2c9c..97f7ee0f04d 100644 --- a/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.h +++ b/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.h @@ -39,12 +39,12 @@ #include "src/trace_processor/core/plugin/registration.h" #include "src/trace_processor/perfetto_sql/engine/dataframe_module.h" #include "src/trace_processor/perfetto_sql/engine/perfetto_sql_database.h" -#include "src/trace_processor/perfetto_sql/engine/pipeline_module.h" #include "src/trace_processor/perfetto_sql/engine/runtime_table_function.h" #include "src/trace_processor/perfetto_sql/engine/static_table_function_module.h" #include "src/trace_processor/perfetto_sql/parser/function_util.h" #include "src/trace_processor/perfetto_sql/parser/perfetto_sql_parser.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" +#include "src/trace_processor/perfetto_sql/pipeline/physical_plan.h" #include "src/trace_processor/sqlite/bindings/sqlite_module.h" #include "src/trace_processor/sqlite/bindings/sqlite_result.h" #include "src/trace_processor/sqlite/bindings/sqlite_window_function.h" @@ -174,6 +174,10 @@ class PerfettoSqlConnection { base::StatusOr PrepareSqliteStatement( SqlSource sql); + // Loads a plan written by pipeline::SerializePlan, ready to run. + base::StatusOr> LoadPipeline( + std::string_view serialized); + // Registers a virtual table module with the given name. // // |name|: name of the module in SQL. @@ -466,7 +470,7 @@ class PerfettoSqlConnection { base::Status ExecuteCreateMacro(const PerfettoSqlParser::CreateMacro&); base::StatusOr PreparePipeline( - pipeline::LogicalPlan, + const pipeline::LogicalPlan&, const SqlSource&); base::Status ExecuteCreateIndex(const PerfettoSqlParser::CreateIndex&); @@ -591,7 +595,6 @@ class PerfettoSqlConnection { // context class of the module inherits from ModuleStateManagerBase. std::vector virtual_module_state_managers_; - PipelineModule::Context* pipeline_context_ = nullptr; RuntimeTableFunctionModule::Context* runtime_table_fn_context_ = nullptr; StaticTableFunctionModule::Context* static_table_fn_context_ = nullptr; DataframeModule::Context* dataframe_context_ = nullptr; 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 d804cdbc0bb..dbfeeaa637f 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 @@ -849,6 +849,19 @@ class PerfettoSqlConnectionPipelineTest : public PerfettoSqlConnectionTest { return rows; } + // The SQL a pipeline statement is prepared as. + std::string PipelineSql(const std::string& pipeline) { + auto res = connection_->ExecuteUntilLastStatement( + SqlSource::FromExecuteQuery(pipeline)); + PERFETTO_CHECK(res.ok()); + return res->stmt.sql(); + } + + // The FROM clause of `sql`, from its leading space. + static std::string From(const std::string& sql) { + return sql.substr(sql.find(" FROM ")); + } + std::vector ColumnNames(const std::string& sql) { auto res = connection_->ExecuteUntilLastStatement( SqlSource::FromExecuteQuery(sql)); @@ -1011,21 +1024,6 @@ TEST_F(PerfettoSqlConnectionPipelineTest, AReadTableCanBeReplaced) { EXPECT_EQ(rows, 4u); } -// Finalization retires the TEMP table; the next execution drops it safely. -TEST_F(PerfettoSqlConnectionPipelineTest, CompletedPipelineTablesAreDropped) { - ASSERT_TRUE(Rows("FROM tree").ok()); - auto tables = Rows( - "SELECT name FROM sqlite_temp_schema WHERE name GLOB " - "'__intrinsic_pipeline_*'"); - ASSERT_TRUE(tables.ok()) << tables.status().message(); - EXPECT_TRUE(tables->empty()); - auto modules = Rows( - "SELECT name FROM pragma_module_list WHERE name GLOB " - "'__intrinsic_pipeline*'"); - ASSERT_TRUE(modules.ok()) << modules.status().message(); - EXPECT_THAT(*modules, testing::ElementsAre("__intrinsic_pipeline")); -} - TEST_F(PerfettoSqlConnectionPipelineTest, DifferentSchemasAndConcurrentStatements) { auto first = @@ -1048,7 +1046,7 @@ TEST_F(PerfettoSqlConnectionPipelineTest, EXPECT_EQ(rows, 4u); EXPECT_TRUE(second->stmt.status().ok()); // The schema changed after preparing the first statement. Resetting and - // rerunning it must retain the bound plan through SQLite's reprepare path. + // rerunning it loads the pipeline again from its plan. ASSERT_EQ(sqlite3_reset(first->stmt.sqlite_stmt()), SQLITE_OK); rows = 0; while (first->stmt.Step()) @@ -1057,54 +1055,32 @@ TEST_F(PerfettoSqlConnectionPipelineTest, EXPECT_TRUE(first->stmt.status().ok()); } -TEST_F(PerfettoSqlConnectionPipelineTest, - CleanupRetriesWhileAnotherStatementIsActive) { - { - auto active = connection_->ExecuteUntilLastStatement( - SqlSource::FromExecuteQuery("FROM tree")); - ASSERT_TRUE(active.ok()) << active.status().message(); - ASSERT_TRUE(Rows("FROM (SELECT 123 AS value)").ok()); - // The finished pipeline can be retired even though the active statement - // prevents schema changes. Retrying cleanup must not interrupt either. - ASSERT_TRUE(Rows("SELECT 1").ok()); - while (active->stmt.Step()) { - } - EXPECT_TRUE(active->stmt.status().ok()); - } - auto tables = Rows( - "SELECT name FROM sqlite_temp_schema WHERE name GLOB " - "'__intrinsic_pipeline_*'"); - ASSERT_TRUE(tables.ok()) << tables.status().message(); - EXPECT_TRUE(tables->empty()); -} +// SQL holding a pipeline's plan can be stored and run later, against the +// tables as they are by then. +TEST_F(PerfettoSqlConnectionPipelineTest, StoredPipelinesRunLater) { + std::string sql = PipelineSql( + "FROM tree |> TREE ACCUMULATE UP SUM(self) AS total |> SELECT id, total"); + ASSERT_TRUE(Rows("CREATE VIEW totals AS " + sql).ok()); + auto rows = Rows("SELECT id, total FROM totals"); + ASSERT_TRUE(rows.ok()) << rows.status().message(); + EXPECT_THAT(*rows, testing::ElementsAre("0,100", "1,60", "2,30", "3,40")); -TEST_F(PerfettoSqlConnectionPipelineTest, - CleanupAcrossRollbackAndExecutionFailure) { - ASSERT_TRUE(Rows("FROM tree").ok()); - ASSERT_TRUE(Rows("BEGIN; FROM tree").ok()); - ASSERT_TRUE(Rows("ROLLBACK").ok()); - EXPECT_FALSE(Rows("FROM (SELECT 0 AS id, NULL AS parent_id, 'bad' AS value) " - "|> TREE ACCUMULATE UP SUM(value) AS total") - .ok()); - auto tables = Rows( - "SELECT name FROM sqlite_temp_schema WHERE name GLOB " - "'__intrinsic_pipeline_*'"); - ASSERT_TRUE(tables.ok()) << tables.status().message(); - EXPECT_TRUE(tables->empty()); + ASSERT_TRUE(Rows("DELETE FROM tree WHERE id = 3").ok()); + rows = Rows("SELECT id, total FROM totals"); + ASSERT_TRUE(rows.ok()) << rows.status().message(); + EXPECT_THAT(*rows, testing::ElementsAre("0,60", "1,20", "2,30")); } -TEST_F(PerfettoSqlConnectionPipelineTest, - FailedCreateDoesNotRetireAnExistingTable) { - ASSERT_TRUE(Rows("CREATE TEMP TABLE __intrinsic_pipeline_0(value); " - "INSERT INTO __intrinsic_pipeline_0 VALUES(123)") - .ok()); - EXPECT_FALSE(Rows("FROM tree").ok()); - auto existing = Rows("SELECT value FROM temp.__intrinsic_pipeline_0"); - ASSERT_TRUE(existing.ok()) << existing.status().message(); - EXPECT_THAT(*existing, testing::ElementsAre("123")); +TEST_F(PerfettoSqlConnectionPipelineTest, BadPlans) { + auto rows = Rows("SELECT c0 FROM __intrinsic_pipeline('FROM tree')"); + EXPECT_THAT(rows.status().message(), testing::HasSubstr("expected a plan")); + rows = Rows("SELECT c0 FROM __intrinsic_pipeline(X'00')"); + EXPECT_THAT(rows.status().message(), testing::HasSubstr("malformed plan")); + rows = Rows("SELECT c1" + From(PipelineSql("FROM (SELECT 1 AS x)"))); + EXPECT_THAT(rows.status().message(), testing::HasSubstr("no such column")); } -TEST_F(PerfettoSqlConnectionPipelineTest, TemporaryTablesAreConnectionLocal) { +TEST_F(PerfettoSqlConnectionPipelineTest, ForksRunPipelinesIndependently) { auto fork = connection_->Fork(); auto first = connection_->ExecuteUntilLastStatement( SqlSource::FromExecuteQuery("FROM (SELECT 123 AS value)")); @@ -1118,55 +1094,14 @@ TEST_F(PerfettoSqlConnectionPipelineTest, TemporaryTablesAreConnectionLocal) { EXPECT_EQ(second->stats.column_count, 2u); } -TEST_F(PerfettoSqlConnectionPipelineTest, - OutputConstraintsApplyAfterAccumulation) { - auto context = std::make_unique(); - context->pool = &pool_; - auto* ctx = context.get(); - connection_->RegisterVirtualTableModule("test_pipeline", - std::move(context)); - ASSERT_TRUE( - Rows("CREATE VIRTUAL TABLE temp.test_output USING test_pipeline(4)") - .ok()); - - pipeline::LogicalPlan logical; - for (const char* name : {"id", "parent_id", "self"}) { - auto id = logical.AddColumn(name, core::Int64{}); - logical.output.push_back({name, id}); - } - pipeline::PlanNodeId scan = logical.AddNode(pipeline::op::Scan{ - SqlSource::FromExecuteQuery("SELECT id, parent_id, self FROM tree"), - logical.output}); - auto total = logical.AddColumn("total", core::Int64{}); - pipeline::op::TreeAccumulate fold; - fold.direction = pipeline::op::TreeDirection::kUp; - fold.node_column = 0; - fold.parent_column = 1; - fold.aggregates.push_back( - {pipeline::op::TreeAccumulate::Function::kSum, 2, total}); - logical.AddNode(std::move(fold), {scan}); - logical.output.push_back({"total", total}); - pipeline::LowerEnvironment env{connection_->sqlite_connection(), &pool_}; - // The root is last in child-first output. Filtering by its output rowid - // must retain all descendants while calculating its total. - for (const char* rhs : {"3", "3.0", "'3'"}) { - auto stmt = connection_->sqlite_connection()->PrepareStatement( - SqlSource::FromExecuteQuery( - "SELECT c3 FROM temp.test_output(?) WHERE rowid = " + - std::string(rhs) + " AND c3 > 50")); - ASSERT_TRUE(stmt.status().ok()) << stmt.status().message(); - PipelineModule::Invocation invocation{ctx, "unused", - pipeline::Lower(logical, env)}; - ASSERT_EQ(sqlite3_bind_pointer(stmt.sqlite_stmt(), 1, &invocation, - PipelineModule::kPlanPointerType, nullptr), - SQLITE_OK); - ASSERT_TRUE(stmt.Step()) << stmt.status().message(); - EXPECT_EQ(sqlite3_column_int64(stmt.sqlite_stmt(), 0), 100); - EXPECT_FALSE(stmt.Step()); - EXPECT_TRUE(stmt.status().ok()); - // Explicitly close cursors before the borrowed plan goes out of scope. - sqlite3_reset(stmt.sqlite_stmt()); - } +// A position in one run of a pipeline need not be the same row in another, +// filtered differently, so its rows have no rowid rather than a misleading one. +TEST_F(PerfettoSqlConnectionPipelineTest, RowsHaveNoRowid) { + std::string from = From(PipelineSql("FROM tree")); + EXPECT_THAT(Rows("SELECT rowid" + from).status().message(), + testing::HasSubstr("no rowid")); + EXPECT_THAT(Rows("SELECT c0" + from + " WHERE rowid = 1").status().message(), + testing::HasSubstr("no rowid")); } TEST_F(PerfettoSqlConnectionPipelineTest, ColumnReadersRefreshAcrossBatches) { diff --git a/src/trace_processor/perfetto_sql/engine/pipeline_module.cc b/src/trace_processor/perfetto_sql/engine/pipeline_module.cc index 9a0ee77ad13..726f253f355 100644 --- a/src/trace_processor/perfetto_sql/engine/pipeline_module.cc +++ b/src/trace_processor/perfetto_sql/engine/pipeline_module.cc @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -28,7 +29,6 @@ #include "perfetto/base/compiler.h" #include "perfetto/base/logging.h" #include "perfetto/base/status.h" -#include "perfetto/ext/base/status_macros.h" #include "perfetto/ext/base/string_utils.h" #include "src/trace_processor/containers/string_pool.h" #include "src/trace_processor/core/common/storage_types.h" @@ -36,7 +36,10 @@ #include "src/trace_processor/core/exec/row_cursor.h" #include "src/trace_processor/core/exec/variant.h" #include "src/trace_processor/core/util/bit_vector.h" +#include "src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.h" +#include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" #include "src/trace_processor/perfetto_sql/pipeline/physical_plan.h" +#include "src/trace_processor/perfetto_sql/pipeline/plan_serialization.h" #include "src/trace_processor/sqlite/bindings/sqlite_result.h" #include "src/trace_processor/sqlite/sqlite_utils.h" @@ -46,13 +49,17 @@ namespace { using core::exec::ColumnView; using core::exec::Variant; -std::string CreateTableStmt(uint32_t column_count) { +// The plan comes before the outputs: SQLite only says which of a table's +// first 63 columns a query reads, and those are best spent on the outputs. +constexpr int kPlanColumn = 0; +constexpr int kFirstOutputColumn = 1; + +std::string Schema() { // Public names (which may repeat) are applied by the outer SELECT. - std::vector columns; - for (uint32_t i = 0; i < column_count; ++i) { + std::vector columns{"pipeline HIDDEN"}; + for (uint32_t i = 0; i < PipelineModule::kMaxColumns; ++i) { columns.push_back("c" + std::to_string(i)); } - columns.emplace_back("plan HIDDEN"); return "CREATE TABLE x(" + base::Join(columns, ", ") + ")"; } @@ -121,6 +128,20 @@ PipelineModule::Cursor::ResultFn FlatReader(const ColumnView& view) { return view.validity() ? &ResultFlat : &ResultFlat; } +void ResultNull(sqlite3_context* ctx, + StringPool*, + const ColumnView&, + uint32_t) { + sqlite::result::Null(ctx); +} + +void ResultNoColumn(sqlite3_context* ctx, + StringPool*, + const ColumnView&, + uint32_t) { + sqlite::utils::SetError(ctx, "__intrinsic_pipeline: no such column"); +} + PipelineModule::Cursor::ResultFn ReaderFor(const ColumnView& view) { if (view.kind() == ColumnView::Kind::kVariant) { return &ResultVariant; @@ -144,10 +165,10 @@ PipelineModule::Cursor::ResultFn ReaderFor(const ColumnView& view) { // Refreshes per-column readers only when entering a new batch. void CacheColumnReaders(PipelineModule::Cursor* c) { - c->columns.clear(); - for (const auto& column : c->plan->columns()) { - const auto& view = c->rows->batch().column(column.index); - c->columns.push_back({&view, ReaderFor(view)}); + const auto& columns = c->plan->columns(); + for (uint32_t i = 0; i < columns.size(); ++i) { + const auto& view = c->rows->batch().column(columns[i].index); + c->columns[kFirstOutputColumn + i] = {&view, ReaderFor(view)}; } } @@ -159,143 +180,92 @@ int CheckStatus(PipelineModule::Cursor* cursor) { : sqlite::utils::SetError(cursor->pVtab, status); } -} // namespace - -PipelineModule::Invocation::~Invocation() { - context->retired_tables.push_back(std::move(table)); -} - -base::Status PipelineModule::Context::Cleanup(sqlite3* db) { - // Otherwise a rollback could resurrect a table after we removed its entry. - if (!sqlite3_get_autocommit(db)) - return base::OkStatus(); - while (!retired_tables.empty()) { - std::string sql = "DROP TABLE IF EXISTS temp." + retired_tables.back(); - int rc = sqlite3_exec(db, sql.c_str(), nullptr, nullptr, nullptr); - if (rc == SQLITE_LOCKED || rc == SQLITE_BUSY) - return base::OkStatus(); - if (rc != SQLITE_OK) - return base::ErrStatus("%s", sqlite3_errmsg(db)); - retired_tables.pop_back(); +// The slow path of Filter: loads the plan in `value` into `c`. +PERFETTO_NO_INLINE int Load(PipelineModule::Cursor* c, sqlite3_value* value) { + if (sqlite3_value_type(value) != SQLITE_BLOB) { + return sqlite::utils::SetError(c->pVtab, + "__intrinsic_pipeline: expected a plan"); } - return base::OkStatus(); + PipelineModule::Context* context = PipelineModule::GetVtab(c->pVtab)->context; + auto plan = context->connection->LoadPipeline( + std::string_view(static_cast(sqlite3_value_blob(value)), + static_cast(sqlite3_value_bytes(value)))); + if (!plan.ok()) { + return sqlite::utils::SetError(c->pVtab, plan.status()); + } + c->plan = std::move(*plan); + c->pool = context->pool; + c->rows = std::make_unique(c->plan->source()); + // One reader per declared column, so Column only indexes: arguments read as + // null, and columns past the plan's outputs fail. + c->columns.assign(kFirstOutputColumn + PipelineModule::kMaxColumns, + {&c->no_view, &ResultNoColumn}); + for (int i = 0; i < kFirstOutputColumn; ++i) { + c->columns[static_cast(i)] = {&c->no_view, &ResultNull}; + } + return SQLITE_OK; } -base::StatusOr PipelineModule::Prepare( - SqliteConnection* connection, - Context* context, - std::unique_ptr plan, - const SqlSource& source) { - RETURN_IF_ERROR(context->Cleanup(connection->db())); - std::string table_name = - "__intrinsic_pipeline_" + std::to_string(context->next_table++); - std::string table = "temp." + table_name; - { - auto create = connection->PrepareStatement( - SqlSource::FromTraceProcessorImplementation( - "CREATE VIRTUAL TABLE " + table + " USING " + kName + "(" + - std::to_string(plan->columns().size()) + ")")); - RETURN_IF_ERROR(create.status()); - create.Step(); - RETURN_IF_ERROR(create.status()); +} // namespace + +base::StatusOr PipelineModule::SelectFrom( + const pipeline::LogicalPlan& plan) { + const std::vector& output = plan.output; + if (output.size() > kMaxColumns) { + return base::ErrStatus("A pipeline can output at most %u columns, not %zu", + kMaxColumns, output.size()); } - auto invocation = std::make_unique(); - invocation->context = context; - invocation->table = std::move(table_name); - invocation->plan = std::move(plan); std::vector columns; - for (size_t i = 0; i < invocation->plan->columns().size(); ++i) { - std::string name = - base::ReplaceAll(invocation->plan->columns()[i].name, "\"", "\"\""); - columns.push_back("c" + std::to_string(i) + " AS \"" + name + "\""); + for (uint32_t i = 0; i < output.size(); ++i) { + columns.push_back("c" + std::to_string(i) + " AS \"" + + base::ReplaceAll(output[i].name, "\"", "\"\"") + "\""); } - auto stmt = connection->PrepareStatement(source.RewriteAllIgnoreExisting( - SqlSource::FromTraceProcessorImplementation( - "SELECT " + base::Join(columns, ", ") + " FROM " + table + "(?)"))); - RETURN_IF_ERROR(stmt.status()); - // Finalization retires the table, so it can be dropped straight after. A - // failure here leaves the table retired and is reported by the next Prepare. - stmt.SetOnFinalized([context, db = connection->db()] { - base::ignore_result(context->Cleanup(db)); - }); - // The hidden plan argument is consumed by xFilter. SQLite releases the plan - // on finalization (also if binding fails), without a generic KeepAlive hook. - int rc = sqlite3_bind_pointer( - stmt.sqlite_stmt(), 1, invocation.release(), kPlanPointerType, - [](void* p) { delete static_cast(p); }); - if (rc != SQLITE_OK) - return base::ErrStatus("%s", sqlite3_errmsg(connection->db())); - return std::move(stmt); + return "SELECT " + base::Join(columns, ", ") + " FROM " + kName + "(X'" + + base::ToHex(pipeline::SerializePlan(plan)) + "')"; } -int PipelineModule::Create(sqlite3* db, - void* raw_ctx, - int argc, - const char* const* argv, - sqlite3_vtab** vtab, - char** error) { - if (argc != 4 || std::string(argv[1]) != "temp") { - *error = sqlite3_mprintf("pipeline tables require TEMP and a column count"); - return SQLITE_ERROR; - } - auto count = base::CStringToUInt32(argv[3]); - if (!count || *count == 0) { - *error = sqlite3_mprintf("invalid pipeline column count"); - return SQLITE_ERROR; - } - std::string create_stmt = CreateTableStmt(*count); - if (int r = sqlite3_declare_vtab(db, create_stmt.c_str()); r != SQLITE_OK) +int PipelineModule::Connect(sqlite3* db, + void* raw_ctx, + int, + const char* const*, + sqlite3_vtab** vtab, + char**) { + std::string schema = Schema(); + if (int r = sqlite3_declare_vtab(db, schema.c_str()); r != SQLITE_OK) { return r; + } auto res = std::make_unique(); res->context = GetContext(raw_ctx); - res->column_count = *count; *vtab = res.release(); return SQLITE_OK; } -int PipelineModule::Connect(sqlite3* db, - void* raw_ctx, - int argc, - const char* const* argv, - sqlite3_vtab** vtab, - char** error) { - return Create(db, raw_ctx, argc, argv, vtab, error); -} - int PipelineModule::Disconnect(sqlite3_vtab* vtab) { std::unique_ptr v(GetVtab(vtab)); return SQLITE_OK; } -int PipelineModule::Destroy(sqlite3_vtab* vtab) { - return Disconnect(vtab); -} - -int PipelineModule::BestIndex(sqlite3_vtab* tab, sqlite3_index_info* info) { +int PipelineModule::BestIndex(sqlite3_vtab*, sqlite3_index_info* info) { int plan = -1; - int rowid = -1; for (int i = 0; i < info->nConstraint; ++i) { const auto& constraint = info->aConstraint[i]; - if (!constraint.usable || constraint.op != SQLITE_INDEX_CONSTRAINT_EQ) + if (constraint.op != SQLITE_INDEX_CONSTRAINT_EQ) { continue; - if (constraint.iColumn == static_cast(GetVtab(tab)->column_count)) + } + if (constraint.iColumn == kPlanColumn) { + // Without the plan there is nothing to run. + if (!constraint.usable) { + return SQLITE_CONSTRAINT; + } plan = i; - if (constraint.iColumn == -1) - rowid = i; + } } - if (plan == -1) + if (plan == -1) { return SQLITE_CONSTRAINT; + } info->aConstraintUsage[plan].argvIndex = 1; info->aConstraintUsage[plan].omit = true; - if (rowid != -1) { - info->aConstraintUsage[rowid].argvIndex = 2; - // SQLite rechecks comparisons, including non-integer RHS values. - info->idxNum = 1; - info->estimatedRows = 1; - } - // Output rowid constraints run after the fold. General predicates remain - // with SQLite; moving them below accumulation could change ancestor totals. - info->estimatedCost = rowid == -1 ? 1e9 : 1e6; + info->estimatedCost = 1e9; return SQLITE_OK; } @@ -310,39 +280,19 @@ int PipelineModule::Close(sqlite3_vtab_cursor* cursor) { } int PipelineModule::Filter(sqlite3_vtab_cursor* cursor, - int idx_num, + int, const char*, int argc, sqlite3_value** argv) { Cursor* c = GetCursor(cursor); - PERFETTO_DCHECK(argc == (idx_num ? 2 : 1)); - auto* invocation = static_cast( - sqlite3_value_pointer(argv[0], kPlanPointerType)); - Vtab* vtab = GetVtab(cursor->pVtab); - if (!invocation || invocation->context != vtab->context || - invocation->plan->columns().size() != vtab->column_count) { - return sqlite::utils::SetError(cursor->pVtab, - "pipeline requires a matching bound plan"); - } - c->plan = invocation->plan.get(); - c->pool = invocation->context->pool; - c->rows = std::make_unique(c->plan->source()); - c->rowid = 0; - c->target_rowid.reset(); - if (idx_num && sqlite3_value_type(argv[1]) == SQLITE_INTEGER) { - c->target_rowid = sqlite3_value_int64(argv[1]); - if (*c->target_rowid < 0) { - c->eof = true; - return SQLITE_OK; + PERFETTO_DCHECK(argc == 1); + // The plan is a constant, so it is loaded once per cursor. + if (PERFETTO_UNLIKELY(!c->plan)) { + if (int rc = Load(c, argv[0]); rc != SQLITE_OK) { + return rc; } } c->eof = !c->rows->Open(); - if (c->target_rowid) { - while (!c->eof && c->rowid < *c->target_rowid) { - ++c->rowid; - c->eof = !c->rows->Next(); - } - } if (c->eof) return CheckStatus(c); CacheColumnReaders(c); @@ -351,11 +301,6 @@ int PipelineModule::Filter(sqlite3_vtab_cursor* cursor, int PipelineModule::Next(sqlite3_vtab_cursor* cursor) { Cursor* c = GetCursor(cursor); - if (c->target_rowid) { - c->eof = true; - return SQLITE_OK; - } - ++c->rowid; c->eof = !c->rows->Next(); if (c->eof) return CheckStatus(c); @@ -370,21 +315,19 @@ int PipelineModule::Eof(sqlite3_vtab_cursor* cursor) { int PipelineModule::Column(sqlite3_vtab_cursor* cursor, sqlite3_context* ctx, - int raw_n) { + int n) { Cursor* c = GetCursor(cursor); - // The hidden argument is a SQL NULL outside xFilter. - if (static_cast(raw_n) == c->columns.size()) { - sqlite::result::Null(ctx); - return SQLITE_OK; - } - const auto& column = c->columns[static_cast(raw_n)]; + const auto& column = c->columns[static_cast(n)]; column.result(ctx, c->pool, *column.view, c->rows->row()); return SQLITE_OK; } -int PipelineModule::Rowid(sqlite3_vtab_cursor* cursor, sqlite_int64* rowid) { - *rowid = GetCursor(cursor)->rowid; - return SQLITE_OK; +int PipelineModule::Rowid(sqlite3_vtab_cursor* cursor, sqlite_int64*) { + // A position in one run need not be the same row in another, filtered + // differently, so rows have no rowid rather than a misleading one. + return sqlite::utils::SetError(cursor->pVtab, + "__intrinsic_pipeline: a pipeline's rows have " + "no rowid"); } } // namespace perfetto::trace_processor diff --git a/src/trace_processor/perfetto_sql/engine/pipeline_module.h b/src/trace_processor/perfetto_sql/engine/pipeline_module.h index b0fe3c064bd..4c7b1961142 100644 --- a/src/trace_processor/perfetto_sql/engine/pipeline_module.h +++ b/src/trace_processor/perfetto_sql/engine/pipeline_module.h @@ -21,49 +21,39 @@ #include #include -#include #include #include #include "perfetto/ext/base/status_or.h" #include "src/trace_processor/containers/string_pool.h" #include "src/trace_processor/core/exec/row_cursor.h" +#include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" #include "src/trace_processor/perfetto_sql/pipeline/physical_plan.h" +#include "src/trace_processor/perfetto_sql/pipeline/plan_serialization.h" #include "src/trace_processor/sqlite/bindings/sqlite_module.h" -#include "src/trace_processor/sqlite/sqlite_connection.h" namespace perfetto::trace_processor { -// One module per connection, with a TEMP virtual table for each pipeline's -// schema. The execution plan is a typed pointer bound to that table's hidden -// argument; SQLite owns it for the lifetime of the prepared statement. +class PerfettoSqlConnection; + +// Runs a pipeline from its plan, serialized into the SQL which reads it: +// `__intrinsic_pipeline(X'...')`. The plan is all a pipeline needs, so the SQL +// can be stored, in a view say, and run later. Pipelines output into fixed +// columns `c0`, `c1`, ..., which the SQL reading them renames. struct PipelineModule : sqlite::Module { - static constexpr auto kType = kCreateOnly; + static constexpr auto kType = kEponymousOnly; static constexpr bool kSupportsWrites = false; static constexpr bool kDoesOverloadFunctions = false; static constexpr char kName[] = "__intrinsic_pipeline"; - static constexpr char kPlanPointerType[] = "perfetto_pipeline_plan"; + static constexpr uint32_t kMaxColumns = pipeline::kMaxPipelineColumns; struct Context { - StringPool* pool; - uint64_t next_table = 0; - // Binding destructors cannot safely perform schema changes. Retire tables - // there and drop them once the pipeline's statement has been finalized. - // Tables which are locked or inside a transaction at that point are - // retried when the next pipeline is finalized or prepared. - std::vector retired_tables; - - base::Status Cleanup(sqlite3*); - }; - struct Invocation { - Context* context; - std::string table; - std::unique_ptr plan; - ~Invocation(); + StringPool* pool = nullptr; + // Loads the plans this module runs. + PerfettoSqlConnection* connection = nullptr; }; struct Vtab : sqlite::Module::Vtab { - Context* context; - uint32_t column_count; + Context* context = nullptr; }; struct Cursor : sqlite::Module::Cursor { using ResultFn = void (*)(sqlite3_context*, @@ -71,31 +61,22 @@ struct PipelineModule : sqlite::Module { const core::exec::ColumnView&, uint32_t); struct ColumnReader { - const core::exec::ColumnView* view; - ResultFn result; + const core::exec::ColumnView* view = nullptr; + ResultFn result = nullptr; }; - const pipeline::PhysicalPlan* plan = nullptr; + std::unique_ptr plan; StringPool* pool = nullptr; std::unique_ptr rows; + // By declared column. std::vector columns; + // What readers of columns with no view are given. + core::exec::ColumnView no_view; bool eof = true; - int64_t rowid = 0; - // An output rowid lookup, applied after every tree fold. - std::optional target_rowid; }; - static base::StatusOr Prepare( - SqliteConnection*, - Context*, - std::unique_ptr, - const SqlSource&); + // SQL reading `plan`'s output under its own column names. + static base::StatusOr SelectFrom(const pipeline::LogicalPlan&); - static int Create(sqlite3*, - void*, - int, - const char* const*, - sqlite3_vtab**, - char**); static int Connect(sqlite3*, void*, int, @@ -103,7 +84,6 @@ struct PipelineModule : sqlite::Module { sqlite3_vtab**, char**); static int Disconnect(sqlite3_vtab*); - static int Destroy(sqlite3_vtab*); static int BestIndex(sqlite3_vtab*, sqlite3_index_info*); diff --git a/src/trace_processor/perfetto_sql/pipeline/BUILD.gn b/src/trace_processor/perfetto_sql/pipeline/BUILD.gn index 16a43632bfc..afa4fdc9eb0 100644 --- a/src/trace_processor/perfetto_sql/pipeline/BUILD.gn +++ b/src/trace_processor/perfetto_sql/pipeline/BUILD.gn @@ -26,6 +26,8 @@ source_set("logical") { "compiler.cc", "compiler.h", "logical_plan.h", + "plan_serialization.cc", + "plan_serialization.h", ] deps = [ "../../../../gn:default_deps", @@ -87,11 +89,15 @@ source_set("test_utils") { perfetto_unittest_source_set("unittests") { testonly = true - sources = [ "physical_plan_unittest.cc" ] + sources = [ + "physical_plan_unittest.cc", + "plan_serialization_unittest.cc", + ] deps = [ ":logical", ":plan", ":test_catalog", + ":test_utils", "../../../../gn:default_deps", "../../../../gn:gtest_and_gmock", "../../../base", diff --git a/src/trace_processor/perfetto_sql/pipeline/physical_plan_unittest.cc b/src/trace_processor/perfetto_sql/pipeline/physical_plan_unittest.cc index 8b0d9d8273f..566c4711003 100644 --- a/src/trace_processor/perfetto_sql/pipeline/physical_plan_unittest.cc +++ b/src/trace_processor/perfetto_sql/pipeline/physical_plan_unittest.cc @@ -28,6 +28,7 @@ #include "perfetto/base/logging.h" #include "perfetto/base/status.h" #include "perfetto/ext/base/flat_hash_map.h" +#include "perfetto/ext/base/status_macros.h" #include "perfetto/ext/base/status_or.h" #include "src/trace_processor/containers/string_pool.h" #include "src/trace_processor/core/common/storage_types.h" @@ -36,6 +37,7 @@ #include "src/trace_processor/core/exec/variant.h" #include "src/trace_processor/perfetto_sql/parser/perfetto_sql_parser.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" +#include "src/trace_processor/perfetto_sql/pipeline/plan_serialization.h" #include "src/trace_processor/perfetto_sql/pipeline/test_catalog.h" #include "src/trace_processor/sqlite/sql_source.h" #include "src/trace_processor/sqlite/sqlite_connection.h" @@ -110,17 +112,22 @@ class PhysicalPlanTest : public ::testing::Test { {{3, 1, 40}, {1, 0, 20}, {2, 0, 30}, {0, std::nullopt, 10}}); } - base::StatusOr> Plan(const std::string& sql) { + base::StatusOr Compile(const std::string& sql) { PerfettoSqlParser parser(macros_, catalog_, /*pipelines_allowed=*/true); parser.Reset(SqlSource::FromExecuteQuery(sql)); if (!parser.Next()) { return parser.status(); } - const auto* pipeline = - std::get_if(&parser.statement()); - PERFETTO_CHECK(pipeline); - return Lower(pipeline->plan, env_); + PERFETTO_CHECK(std::holds_alternative( + parser.statement())); + return std::move( + std::get(parser.TakeStatement()).plan); + } + + base::StatusOr> Plan(const std::string& sql) { + ASSIGN_OR_RETURN(LogicalPlan plan, Compile(sql)); + return Lower(plan, env_); } std::vector Names(const PhysicalPlan& plan) { @@ -386,5 +393,22 @@ TEST_F(PhysicalPlanTest, APlanCanRunMoreThanOnce) { EXPECT_EQ(*first, *second); } +// A plan read back from its bytes runs, reading a table replaced by one of +// the same shape as it is now. +TEST_F(PhysicalPlanTest, APlanReadBackReadsTablesAsTheyAreNow) { + CreateDataframeTree(); + auto plan = Compile("FROM df |> TREE ACCUMULATE UP SUM(self) AS total"); + ASSERT_TRUE(plan.ok()) << plan.status().message(); + std::string bytes = SerializePlan(*plan); + catalog_.RemoveTable("df"); + catalog_.AddTable("df", {"id", "parent_id", "self"}, + {{1, 0, 2}, {0, std::nullopt, 1}}); + auto read = DeserializePlan(bytes, catalog_); + ASSERT_TRUE(read.ok()) << read.status().message(); + auto rows = Run(*Lower(*read, env_), "total"); + ASSERT_TRUE(rows.ok()) << rows.status().message(); + EXPECT_THAT(*rows, ElementsAre(Pair(0, 3), Pair(1, 2))); +} + } // namespace } // namespace perfetto::trace_processor::pipeline diff --git a/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc b/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc new file mode 100644 index 00000000000..fb5eacd4cb6 --- /dev/null +++ b/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc @@ -0,0 +1,438 @@ +/* + * 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/perfetto_sql/pipeline/plan_serialization.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "perfetto/base/logging.h" +#include "perfetto/base/status.h" +#include "perfetto/ext/base/status_macros.h" +#include "perfetto/ext/base/status_or.h" +#include "src/trace_processor/core/common/storage_types.h" +#include "src/trace_processor/core/dataframe/dataframe.h" +#include "src/trace_processor/core/dataframe/specs.h" +#include "src/trace_processor/perfetto_sql/pipeline/catalog.h" +#include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" +#include "src/trace_processor/sqlite/sql_source.h" + +namespace perfetto::trace_processor::pipeline { +namespace { + +// A plan is written as its source then the stages above it, naming columns by +// position among those available, so only plans the compiler could build can +// be written and reading one needs only bounds checks. + +// The columns available to a stage, in the order positions number them. +using Available = std::vector; + +class Writer { + public: + void U8(uint8_t value) { Append(&value, sizeof(value)); } + void U32(uint32_t value) { Append(&value, sizeof(value)); } + void Size(size_t value) { U32(static_cast(value)); } + void Str(std::string_view value) { + Size(value.size()); + out_.append(value.data(), value.size()); + } + // Out of bounds if `id` is not available. + void Position(const Available& available, ColumnId id) { + auto it = std::find(available.begin(), available.end(), id); + Size(static_cast(it - available.begin())); + } + std::string Take() { return std::move(out_); } + + private: + void Append(const void* data, size_t size) { + size_t at = out_.size(); + out_.resize(at + size); + memcpy(out_.data() + at, data, size); + } + + std::string out_; +}; + +// After the first failure, every read returns zeroes. +class Reader { + public: + explicit Reader(std::string_view in) : in_(in) {} + + uint8_t U8() { + uint8_t value = 0; + Read(&value, sizeof(value)); + return value; + } + uint32_t U32() { + uint32_t value = 0; + Read(&value, sizeof(value)); + return value; + } + // At most the bytes left, so a corrupt count cannot allocate much. + uint32_t Count() { + uint32_t count = U32(); + if (count > in_.size() - at_) { + Fail(); + return 0; + } + return count; + } + std::string Str() { + uint32_t size = Count(); + std::string value(in_.substr(at_, size)); + at_ += size; + return value; + } + ColumnId Position(const Available& available) { + uint32_t i = U32(); + if (i >= available.size()) { + Fail(); + return 0; + } + return available[i]; + } + + void Fail() { ok_ = false; } + bool ok() const { return ok_; } + bool done() const { return ok_ && at_ == in_.size(); } + + private: + void Read(void* out, size_t size) { + if (!ok_ || in_.size() - at_ < size) { + Fail(); + return; + } + memcpy(out, in_.data() + at_, size); + at_ += size; + } + + std::string_view in_; + size_t at_ = 0; + bool ok_ = true; +}; + +void WriteType(Writer& w, const std::optional& type) { + // Zero is a type known only per row; otherwise the type's index plus one. + w.U8(type ? static_cast(type->index() + 1) : 0); +} + +std::optional ReadType(Reader& r) { + uint8_t tag = r.U8(); + if (tag == 0) { + return std::nullopt; + } + switch (tag - 1) { + case core::StorageType::GetTypeIndex(): + return core::Id{}; + case core::StorageType::GetTypeIndex(): + return core::Uint32{}; + case core::StorageType::GetTypeIndex(): + return core::Int32{}; + case core::StorageType::GetTypeIndex(): + return core::Int64{}; + case core::StorageType::GetTypeIndex(): + return core::Double{}; + case core::StorageType::GetTypeIndex(): + return core::String{}; + default: + r.Fail(); + return std::nullopt; + } +} + +class PlanWriter { + public: + explicit PlanWriter(const LogicalPlan& plan) : plan_(plan) {} + + std::string Write() { + // Pruning can leave nodes the root does not reach. + std::vector folds; + PlanNodeId id = plan_.root; + while (plan_.nodes[id].Is()) { + folds.push_back(&plan_.nodes[id].Cast()); + id = plan_.nodes[id].children[0]; + } + const PlanNode& source = plan_.nodes[id]; + w_.U8(static_cast(source.op.index())); + 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(plan_.output.size()); + for (const NamedColumn& column : plan_.output) { + w_.Str(column.name); + w_.Position(available, column.id); + } + return w_.Take(); + } + + private: + Available WriteScan(const op::Scan& scan) { + w_.U8(static_cast(scan.source.index())); + switch (scan.source.index()) { + case base::variant_index(): + w_.Str(base::unchecked_get(scan.source).name); + break; + case base::variant_index(): + w_.Str(base::unchecked_get(scan.source).sql()); + break; + default: + PERFETTO_FATAL("Unknown scan source"); + } + w_.Size(scan.columns.size()); + Available available; + for (const NamedColumn& column : scan.columns) { + w_.Str(column.name); + WriteType(w_, plan_.columns[column.id].type); + available.push_back(column.id); + } + return available; + } + + void WriteTreeAccumulate(const op::TreeAccumulate& acc, + Available& available) { + w_.U8(static_cast(acc.direction)); + w_.Position(available, acc.node_column); + w_.Position(available, acc.parent_column); + w_.Size(acc.aggregates.size()); + Available outputs; + for (const op::TreeAccumulate::Aggregate& agg : acc.aggregates) { + w_.Position(available, agg.column); + w_.Str(plan_.columns[agg.output].name); + outputs.push_back(agg.output); + } + available.insert(available.end(), outputs.begin(), outputs.end()); + } + + Available WriteIntervalIntersect(const PlanNode& node) { + const auto& isect = node.Cast(); + w_.Str(plan_.columns[isect.ts].name); + w_.Str(plan_.columns[isect.dur].name); + w_.Size(isect.operands.size()); + w_.Size(isect.operands.empty() ? 0 : isect.operands[0].keys.size()); + Available available{isect.ts, isect.dur}; + for (uint32_t k = 0; k < isect.operands.size(); ++k) { + const op::IntervalIntersect::Operand& operand = isect.operands[k]; + PERFETTO_CHECK(plan_.nodes[node.children[k]].Is()); + Available in = WriteScan(plan_.nodes[node.children[k]].Cast()); + w_.Position(in, operand.ts); + w_.Position(in, operand.dur); + for (ColumnId id : operand.keys) { + w_.Position(in, id); + } + w_.Size(operand.carried.size()); + for (ColumnId id : operand.carried) { + w_.Position(in, id); + available.push_back(id); + } + } + return available; + } + + const LogicalPlan& plan_; + Writer w_; +}; + +class PlanReader { + public: + explicit PlanReader(Reader& r) : r_(r) {} + + LogicalPlan Read() { + Available available; + switch (r_.U8()) { + case base::variant_index(): { + op::Scan scan; + available = ReadScan(scan); + plan_.AddNode(std::move(scan)); + break; + } + case base::variant_index(): + ReadIntervalIntersect(available); + break; + default: + r_.Fail(); + return {}; + } + uint32_t folds = r_.Count(); + for (uint32_t i = 0; i < folds && r_.ok(); ++i) { + plan_.AddNode(ReadTreeAccumulate(available), {plan_.root}); + } + // The table function declares only so many output columns. + uint32_t outputs = r_.Count(); + if (outputs > kMaxPipelineColumns) { + r_.Fail(); + return {}; + } + plan_.output.resize(outputs); + for (NamedColumn& column : plan_.output) { + column.name = r_.Str(); + column.id = r_.Position(available); + } + return std::move(plan_); + } + + private: + Available ReadScan(op::Scan& scan) { + switch (r_.U8()) { + case base::variant_index(): { + op::Scan::Dataframe source; + source.name = r_.Str(); + scan.source = std::move(source); + break; + } + case base::variant_index(): + scan.source = SqlSource::FromTraceProcessorImplementation(r_.Str()); + break; + default: + r_.Fail(); + return {}; + } + scan.columns.resize(r_.Count()); + bool from_sql = std::holds_alternative(scan.source); + Available available; + for (NamedColumn& column : scan.columns) { + column.name = r_.Str(); + std::optional type = ReadType(r_); + // SqlScan cannot produce Ids. + if (from_sql && type && type->Is()) { + r_.Fail(); + } + column.id = plan_.AddColumn(column.name, type); + available.push_back(column.id); + } + return available; + } + + op::TreeAccumulate ReadTreeAccumulate(Available& available) { + op::TreeAccumulate acc; + acc.direction = static_cast(r_.U8()); + acc.node_column = r_.Position(available); + acc.parent_column = r_.Position(available); + acc.aggregates.resize(r_.Count()); + for (op::TreeAccumulate::Aggregate& agg : acc.aggregates) { + agg.column = r_.Position(available); + agg.output = plan_.AddColumn(r_.Str(), core::Int64{}); + } + for (const op::TreeAccumulate::Aggregate& agg : acc.aggregates) { + available.push_back(agg.output); + } + return acc; + } + + void ReadIntervalIntersect(Available& available) { + op::IntervalIntersect isect; + isect.ts = plan_.AddColumn(r_.Str(), core::Int64{}); + isect.dur = plan_.AddColumn(r_.Str(), core::Int64{}); + available = {isect.ts, isect.dur}; + isect.operands.resize(r_.Count()); + uint32_t keys = r_.Count(); + if (isect.operands.size() < 2) { + r_.Fail(); + } + std::vector children; + for (op::IntervalIntersect::Operand& operand : isect.operands) { + op::Scan scan; + Available in = ReadScan(scan); + if (!r_.ok()) { + return; + } + children.push_back(plan_.AddNode(std::move(scan))); + operand.ts = r_.Position(in); + operand.dur = r_.Position(in); + operand.keys.resize(keys); + for (ColumnId& id : operand.keys) { + id = r_.Position(in); + } + operand.carried.resize(r_.Count()); + for (ColumnId& id : operand.carried) { + id = r_.Position(in); + available.push_back(id); + } + } + plan_.AddNode(std::move(isect), std::move(children)); + } + + Reader& r_; + LogicalPlan plan_; +}; + +// Points each dataframe scan at the dataframe now registered under its name, +// which must still have every column the plan reads, with the same type. +base::Status ResolveDataframes(LogicalPlan& plan, const Catalog& catalog) { + for (PlanNode& node : plan.nodes) { + if (!node.Is()) { + continue; + } + auto& scan = node.Cast(); + auto* source = std::get_if(&scan.source); + if (!source) { + continue; + } + const dataframe::Dataframe* dataframe = catalog.FindDataframe(source->name); + if (!dataframe) { + return base::ErrStatus("Pipeline: table '%s' no longer exists", + source->name.c_str()); + } + const std::vector& names = dataframe->column_names(); + for (const NamedColumn& column : scan.columns) { + uint32_t i = 0; + while (i < names.size() && + (names[i] != column.name || dataframe::IsHiddenColumn(names[i]))) { + ++i; + } + const std::optional& type = + plan.columns[column.id].type; + if (i == names.size() || !type || !(*type == dataframe->column_type(i))) { + return base::ErrStatus( + "Pipeline: table '%s' has changed since the pipeline was written", + source->name.c_str()); + } + source->columns.push_back(dataframe->shared_column(i)); + } + source->row_count = dataframe->row_count(); + } + return base::OkStatus(); +} + +} // namespace + +std::string SerializePlan(const LogicalPlan& plan) { + return PlanWriter(plan).Write(); +} + +base::StatusOr DeserializePlan(std::string_view bytes, + const Catalog& catalog) { + Reader r(bytes); + LogicalPlan plan = PlanReader(r).Read(); + if (!r.done()) { + return base::ErrStatus("__intrinsic_pipeline: malformed plan"); + } + RETURN_IF_ERROR(ResolveDataframes(plan, catalog)); + return std::move(plan); +} + +} // namespace perfetto::trace_processor::pipeline diff --git a/src/trace_processor/perfetto_sql/pipeline/plan_serialization.h b/src/trace_processor/perfetto_sql/pipeline/plan_serialization.h new file mode 100644 index 00000000000..28db0710962 --- /dev/null +++ b/src/trace_processor/perfetto_sql/pipeline/plan_serialization.h @@ -0,0 +1,45 @@ +/* + * 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_PERFETTO_SQL_PIPELINE_PLAN_SERIALIZATION_H_ +#define SRC_TRACE_PROCESSOR_PERFETTO_SQL_PIPELINE_PLAN_SERIALIZATION_H_ + +#include +#include +#include + +#include "perfetto/ext/base/status_or.h" +#include "src/trace_processor/perfetto_sql/pipeline/catalog.h" +#include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" + +namespace perfetto::trace_processor::pipeline { + +// The most columns a pipeline can output. +inline constexpr uint32_t kMaxPipelineColumns = 256; + +// Writes an optimized plan as bytes, so that SQL can carry it and run it later +// without compiling the pipeline again. +std::string SerializePlan(const LogicalPlan&); + +// Rebuilds a plan written by SerializePlan, looking its dataframes up in +// `catalog` again. Anyone can write bytes into SQL, so anything SerializePlan +// could not have written is refused, as is a plan whose dataframes have since +// changed. +base::StatusOr DeserializePlan(std::string_view, const Catalog&); + +} // namespace perfetto::trace_processor::pipeline + +#endif // SRC_TRACE_PROCESSOR_PERFETTO_SQL_PIPELINE_PLAN_SERIALIZATION_H_ diff --git a/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc b/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc new file mode 100644 index 00000000000..564ecb45062 --- /dev/null +++ b/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc @@ -0,0 +1,196 @@ +/* + * 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/perfetto_sql/pipeline/plan_serialization.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "perfetto/base/logging.h" +#include "perfetto/ext/base/flat_hash_map.h" +#include "perfetto/ext/base/status_or.h" +#include "src/trace_processor/containers/string_pool.h" +#include "src/trace_processor/core/exec/row_cursor.h" +#include "src/trace_processor/perfetto_sql/parser/perfetto_sql_parser.h" +#include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" +#include "src/trace_processor/perfetto_sql/pipeline/logical_plan_test_utils.h" +#include "src/trace_processor/perfetto_sql/pipeline/physical_plan.h" +#include "src/trace_processor/perfetto_sql/pipeline/test_catalog.h" +#include "src/trace_processor/sqlite/sql_source.h" +#include "src/trace_processor/sqlite/sqlite_connection.h" +#include "test/gtest_and_gmock.h" + +namespace perfetto::trace_processor::pipeline { +namespace { + +using ::testing::HasSubstr; + +class PlanSerializationTest : public ::testing::Test { + protected: + PlanSerializationTest() + : connection_(SqliteConnection::CreateConnectionToNewDatabase()), + catalog_(&pool_, connection_.get()) { + env_.connection = connection_.get(); + env_.pool = &pool_; + // 0 (10) -> 1 (20) -> 3 (40) + // -> 2 (30) + catalog_.AddTable( + "df", {"id", "parent_id", "self"}, + {{3, 1, 40}, {1, 0, 20}, {2, 0, 30}, {0, std::nullopt, 10}}); + Exec("CREATE TABLE spans(ts INTEGER, dur INTEGER, cpu INTEGER)"); + Exec("INSERT INTO spans VALUES (0, 10, 1), (5, 10, 1), (20, 5, 2)"); + } + + void Exec(const std::string& sql) { + auto statement = + connection_->PrepareStatement(SqlSource::FromExecuteQuery(sql)); + while (statement.Step()) { + } + ASSERT_TRUE(statement.status().ok()) << statement.status().c_message(); + } + + LogicalPlan Compile(const std::string& sql) { + PerfettoSqlParser parser(macros_, catalog_, /*pipelines_allowed=*/true); + parser.Reset(SqlSource::FromExecuteQuery(sql)); + PERFETTO_CHECK(parser.Next()); + return std::move( + std::get(parser.TakeStatement()).plan); + } + + base::StatusOr RoundTrip(const LogicalPlan& plan) { + return DeserializePlan(SerializePlan(plan), catalog_); + } + + StringPool pool_; + std::unique_ptr connection_; + TestCatalog catalog_; + LowerEnvironment env_; + base::FlatHashMap macros_; +}; + +// Reading numbers columns again, so compare plans up to renumbering. +std::string Renumbered(const LogicalPlan& plan) { + std::string in = LogicalPlanToString(plan); + std::string out; + std::map numbers; + for (size_t i = 0; i < in.size();) { + if (in[i] != '#') { + out += in[i++]; + continue; + } + size_t end = in.find_first_not_of("0123456789", i + 1); + auto [it, inserted] = + numbers.emplace(in.substr(i, end - i), numbers.size()); + out += "#" + std::to_string(it->second); + i = end; + } + return out; +} + +const char* const kPipelines[] = { + "FROM df |> TREE ACCUMULATE UP SUM(self) AS total |> SELECT id, total", + "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", + // Pruning removes the fold, whose node is left behind unread. + "FROM df |> TREE ACCUMULATE UP SUM(self) AS total |> SELECT id", +}; + +TEST_F(PlanSerializationTest, EveryOperatorRoundTrips) { + for (const char* sql : kPipelines) { + LogicalPlan plan = Compile(sql); + auto read = RoundTrip(plan); + ASSERT_TRUE(read.ok()) << sql << ": " << read.status().message(); + EXPECT_EQ(Renumbered(*read), Renumbered(plan)) << sql; + } +} + +TEST_F(PlanSerializationTest, ChangedTablesAreRefused) { + std::string bytes = SerializePlan(Compile( + "FROM df |> TREE ACCUMULATE UP SUM(self) AS total |> SELECT id, total")); + catalog_.RemoveTable("df"); + EXPECT_THAT(DeserializePlan(bytes, catalog_).status().message(), + HasSubstr("'df' no longer exists")); + catalog_.AddTable("df", {"id", "parent_id", "other"}, {{0, std::nullopt, 1}}); + EXPECT_THAT(DeserializePlan(bytes, catalog_).status().message(), + HasSubstr("'df' has changed")); +} + +// Anyone can write a plan into SQL, so no bytes may do worse than fail: cut +// short or padded bytes are refused, and corrupt ones are refused or run. +TEST_F(PlanSerializationTest, MalformedPlansAreRefusedOrRun) { + for (const char* sql : kPipelines) { + std::string bytes = SerializePlan(Compile(sql)); + for (size_t size = 0; size < bytes.size(); ++size) { + EXPECT_FALSE(DeserializePlan(bytes.substr(0, size), catalog_).ok()) + << sql << " cut to " << size; + } + EXPECT_FALSE(DeserializePlan(bytes + '\0', catalog_).ok()) << sql; + for (size_t at = 0; at < bytes.size(); ++at) { + for (char value : {'\0', '\1', '\x7f', '\xff'}) { + std::string corrupt = bytes; + corrupt[at] = value; + auto read = DeserializePlan(corrupt, catalog_); + if (!read.ok()) { + continue; + } + auto physical = Lower(*read, env_); + core::exec::RowCursor cursor(physical->source()); + for (bool row = cursor.Open(); row; row = cursor.Next()) { + } + } + } + } +} + +TEST_F(PlanSerializationTest, PlansLoweringCannotRunAreRefused) { + LogicalPlan plan = Compile( + "FROM df |> TREE ACCUMULATE UP SUM(self) AS total |> SELECT id, total"); + ASSERT_EQ(plan.nodes.size(), 2u); + + LogicalPlan unknown_column = plan; + unknown_column.output[0].id = 99; + EXPECT_FALSE(RoundTrip(unknown_column).ok()); + + LogicalPlan reads_nothing_below = plan; + reads_nothing_below.nodes[1].Cast().node_column = + reads_nothing_below.nodes[1] + .Cast() + .aggregates[0] + .output; + EXPECT_FALSE(RoundTrip(reads_nothing_below).ok()); +} + +TEST_F(PlanSerializationTest, PlansWithTooManyOutputsAreRefused) { + LogicalPlan plan = Compile("FROM df |> SELECT id"); + ASSERT_EQ(plan.output.size(), 1u); + + LogicalPlan at_limit = plan; + at_limit.output.assign(kMaxPipelineColumns, plan.output[0]); + EXPECT_TRUE(RoundTrip(at_limit).ok()); + + LogicalPlan over_limit = plan; + over_limit.output.assign(kMaxPipelineColumns + 1, plan.output[0]); + EXPECT_FALSE(RoundTrip(over_limit).ok()); +} + +} // namespace +} // namespace perfetto::trace_processor::pipeline diff --git a/src/trace_processor/sqlite/sqlite_connection.cc b/src/trace_processor/sqlite/sqlite_connection.cc index ba41d5e9ff0..c05f4a585ed 100644 --- a/src/trace_processor/sqlite/sqlite_connection.cc +++ b/src/trace_processor/sqlite/sqlite_connection.cc @@ -274,15 +274,7 @@ void* SqliteConnection::SetRollbackCallback(RollbackCallback callback, SqliteConnection::PreparedStatement::PreparedStatement(ScopedStmt stmt, SqlSource source) - : stmt_(stmt.release()), sql_source_(std::move(source)) {} - -void SqliteConnection::PreparedStatement::Finalizer::operator()( - sqlite3_stmt* stmt) const { - sqlite3_finalize(stmt); - if (on_finalized) { - on_finalized(); - } -} + : stmt_(std::move(stmt)), sql_source_(std::move(source)) {} bool SqliteConnection::PreparedStatement::Step() { PERFETTO_TP_TRACE(metatrace::Category::QUERY_DETAILED, "STMT_STEP", diff --git a/src/trace_processor/sqlite/sqlite_connection.h b/src/trace_processor/sqlite/sqlite_connection.h index ca84de25fb7..64b59536b8f 100644 --- a/src/trace_processor/sqlite/sqlite_connection.h +++ b/src/trace_processor/sqlite/sqlite_connection.h @@ -20,7 +20,6 @@ #include #include #include -#include #include #include #include @@ -72,24 +71,12 @@ class SqliteConnection { const base::Status& status() const { return status_; } sqlite3_stmt* sqlite_stmt() const { return stmt_.get(); } - // Sets a callback to run once the statement has been finalized. Unlike the - // destructors SQLite runs during finalization, it can change the schema. - void SetOnFinalized(std::function fn) { - stmt_.get_deleter().on_finalized = std::move(fn); - } - private: friend class SqliteConnection; - // Finalizes the statement, then runs `on_finalized` if set. - struct Finalizer { - void operator()(sqlite3_stmt*) const; - std::function on_finalized; - }; - explicit PreparedStatement(ScopedStmt, SqlSource); - std::unique_ptr stmt_; + ScopedStmt stmt_; ScopedSqliteString expanded_sql_; SqlSource sql_source_; base::Status status_ = base::OkStatus();