diff --git a/Android.bp b/Android.bp
index 7edd8838e2..862fd6c4d2 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 a5f594afaf..8ecb59b1b5 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 0706771d67..8070ed2180 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<PipelineModule::Context>();
     ctx->pool = pool_;
-    pipeline_context_ = ctx.get();
+    ctx->connection = this;
     RegisterVirtualTableModule<PipelineModule>(PipelineModule::kName,
                                                std::move(ctx));
   }
@@ -701,9 +702,8 @@ PerfettoSqlConnection::ProcessFrame(size_t frame_idx) {
             std::holds_alternative<PerfettoSqlParser::SqliteSql>(stmt))) {
       source_to_prepare = parser->TakeStatementSql();
     } else if (std::holds_alternative<PerfettoSqlParser::Pipeline>(stmt)) {
-      auto pipeline =
-          std::get<PerfettoSqlParser::Pipeline>(parser->TakeStatement());
-      pipeline_plan = std::move(pipeline.plan);
+      pipeline_plan = std::move(
+          std::get<PerfettoSqlParser::Pipeline>(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<pipeline::LogicalPlan>(&create_table.body);
+  const auto* logical = std::get_if<pipeline::LogicalPlan>(&create_table.body);
   base::StatusOr<SqliteConnection::PreparedStatement> stmt_or =
-      logical ? PreparePipeline(std::move(*logical), statement_sql)
+      logical ? PreparePipeline(*logical, statement_sql)
               : connection_->PrepareStatement(
                     std::move(std::get<SqlSource>(create_table.body)));
   ASSIGN_OR_RETURN(auto stmt, std::move(stmt_or));
@@ -1049,14 +1049,26 @@ base::Status PerfettoSqlConnection::ExecuteCreateTable(
 }
 
 base::StatusOr<SqliteConnection::PreparedStatement>
-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<std::unique_ptr<pipeline::PhysicalPlan>>
+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 3b368fd2c9..97f7ee0f04 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<SqliteConnection::PreparedStatement> PrepareSqliteStatement(
       SqlSource sql);
 
+  // Loads a plan written by pipeline::SerializePlan, ready to run.
+  base::StatusOr<std::unique_ptr<pipeline::PhysicalPlan>> 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<SqliteConnection::PreparedStatement> 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<sqlite::ModuleStateManagerBase*> 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 d804cdbc0b..dbfeeaa637 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<std::string> 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<PipelineModule::Context>();
-  context->pool = &pool_;
-  auto* ctx = context.get();
-  connection_->RegisterVirtualTableModule<PipelineModule>("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 9a0ee77ad1..726f253f35 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 <cstdint>
 #include <memory>
 #include <string>
+#include <string_view>
 #include <type_traits>
 #include <utility>
 #include <vector>
@@ -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<std::string> columns;
-  for (uint32_t i = 0; i < column_count; ++i) {
+  std::vector<std::string> 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<T, true> : &ResultFlat<T, false>;
 }
 
+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<const char*>(sqlite3_value_blob(value)),
+                       static_cast<size_t>(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<core::exec::RowCursor>(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<uint32_t>(i)] = {&c->no_view, &ResultNull};
+  }
+  return SQLITE_OK;
 }
 
-base::StatusOr<SqliteConnection::PreparedStatement> PipelineModule::Prepare(
-    SqliteConnection* connection,
-    Context* context,
-    std::unique_ptr<pipeline::PhysicalPlan> 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<std::string> PipelineModule::SelectFrom(
+    const pipeline::LogicalPlan& plan) {
+  const std::vector<pipeline::NamedColumn>& 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>();
-  invocation->context = context;
-  invocation->table = std::move(table_name);
-  invocation->plan = std::move(plan);
   std::vector<std::string> 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<Invocation*>(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<Vtab>();
   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<Vtab> 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<int>(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<Invocation*>(
-      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<core::exec::RowCursor>(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<uint32_t>(raw_n) == c->columns.size()) {
-    sqlite::result::Null(ctx);
-    return SQLITE_OK;
-  }
-  const auto& column = c->columns[static_cast<uint32_t>(raw_n)];
+  const auto& column = c->columns[static_cast<uint32_t>(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 b0fe3c064b..4c7b196114 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 <cstdint>
 #include <memory>
-#include <optional>
 #include <string>
 #include <vector>
 
 #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<PipelineModule> {
-  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<std::string> retired_tables;
-
-    base::Status Cleanup(sqlite3*);
-  };
-  struct Invocation {
-    Context* context;
-    std::string table;
-    std::unique_ptr<pipeline::PhysicalPlan> plan;
-    ~Invocation();
+    StringPool* pool = nullptr;
+    // Loads the plans this module runs.
+    PerfettoSqlConnection* connection = nullptr;
   };
   struct Vtab : sqlite::Module<PipelineModule>::Vtab {
-    Context* context;
-    uint32_t column_count;
+    Context* context = nullptr;
   };
   struct Cursor : sqlite::Module<PipelineModule>::Cursor {
     using ResultFn = void (*)(sqlite3_context*,
@@ -71,31 +61,22 @@ struct PipelineModule : sqlite::Module<PipelineModule> {
                               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<pipeline::PhysicalPlan> plan;
     StringPool* pool = nullptr;
     std::unique_ptr<core::exec::RowCursor> rows;
+    // By declared column.
     std::vector<ColumnReader> 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<int64_t> target_rowid;
   };
 
-  static base::StatusOr<SqliteConnection::PreparedStatement> Prepare(
-      SqliteConnection*,
-      Context*,
-      std::unique_ptr<pipeline::PhysicalPlan>,
-      const SqlSource&);
+  // SQL reading `plan`'s output under its own column names.
+  static base::StatusOr<std::string> 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<PipelineModule> {
                      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 16a43632bf..afa4fdc9eb 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 8b0d9d8273..566c471100 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<std::unique_ptr<PhysicalPlan>> Plan(const std::string& sql) {
+  base::StatusOr<LogicalPlan> 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<PerfettoSqlParser::Pipeline>(&parser.statement());
-    PERFETTO_CHECK(pipeline);
-    return Lower(pipeline->plan, env_);
+    PERFETTO_CHECK(std::holds_alternative<PerfettoSqlParser::Pipeline>(
+        parser.statement()));
+    return std::move(
+        std::get<PerfettoSqlParser::Pipeline>(parser.TakeStatement()).plan);
+  }
+
+  base::StatusOr<std::unique_ptr<PhysicalPlan>> Plan(const std::string& sql) {
+    ASSIGN_OR_RETURN(LogicalPlan plan, Compile(sql));
+    return Lower(plan, env_);
   }
 
   std::vector<std::string> 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 0000000000..fb5eacd4cb
--- /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 <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <variant>
+#include <vector>
+
+#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<ColumnId>;
+
+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<uint32_t>(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<size_t>(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<core::StorageType>& type) {
+  // Zero is a type known only per row; otherwise the type's index plus one.
+  w.U8(type ? static_cast<uint8_t>(type->index() + 1) : 0);
+}
+
+std::optional<core::StorageType> ReadType(Reader& r) {
+  uint8_t tag = r.U8();
+  if (tag == 0) {
+    return std::nullopt;
+  }
+  switch (tag - 1) {
+    case core::StorageType::GetTypeIndex<core::Id>():
+      return core::Id{};
+    case core::StorageType::GetTypeIndex<core::Uint32>():
+      return core::Uint32{};
+    case core::StorageType::GetTypeIndex<core::Int32>():
+      return core::Int32{};
+    case core::StorageType::GetTypeIndex<core::Int64>():
+      return core::Int64{};
+    case core::StorageType::GetTypeIndex<core::Double>():
+      return core::Double{};
+    case core::StorageType::GetTypeIndex<core::String>():
+      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<const op::TreeAccumulate*> folds;
+    PlanNodeId id = plan_.root;
+    while (plan_.nodes[id].Is<op::TreeAccumulate>()) {
+      folds.push_back(&plan_.nodes[id].Cast<op::TreeAccumulate>());
+      id = plan_.nodes[id].children[0];
+    }
+    const PlanNode& source = plan_.nodes[id];
+    w_.U8(static_cast<uint8_t>(source.op.index()));
+    Available available = source.Is<op::Scan>()
+                              ? WriteScan(source.Cast<op::Scan>())
+                              : WriteIntervalIntersect(source);
+    w_.Size(folds.size());
+    for (auto it = folds.rbegin(); it != folds.rend(); ++it) {
+      WriteTreeAccumulate(**it, available);
+    }
+    w_.Size(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<uint8_t>(scan.source.index()));
+    switch (scan.source.index()) {
+      case base::variant_index<op::Scan::Source, op::Scan::Dataframe>():
+        w_.Str(base::unchecked_get<op::Scan::Dataframe>(scan.source).name);
+        break;
+      case base::variant_index<op::Scan::Source, SqlSource>():
+        w_.Str(base::unchecked_get<SqlSource>(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<uint8_t>(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<op::IntervalIntersect>();
+    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<op::Scan>());
+      Available in = WriteScan(plan_.nodes[node.children[k]].Cast<op::Scan>());
+      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, op::Scan>(): {
+        op::Scan scan;
+        available = ReadScan(scan);
+        plan_.AddNode(std::move(scan));
+        break;
+      }
+      case base::variant_index<Op, op::IntervalIntersect>():
+        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::Source, op::Scan::Dataframe>(): {
+        op::Scan::Dataframe source;
+        source.name = r_.Str();
+        scan.source = std::move(source);
+        break;
+      }
+      case base::variant_index<op::Scan::Source, SqlSource>():
+        scan.source = SqlSource::FromTraceProcessorImplementation(r_.Str());
+        break;
+      default:
+        r_.Fail();
+        return {};
+    }
+    scan.columns.resize(r_.Count());
+    bool from_sql = std::holds_alternative<SqlSource>(scan.source);
+    Available available;
+    for (NamedColumn& column : scan.columns) {
+      column.name = r_.Str();
+      std::optional<core::StorageType> type = ReadType(r_);
+      // SqlScan cannot produce Ids.
+      if (from_sql && type && type->Is<core::Id>()) {
+        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<op::TreeDirection>(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<PlanNodeId> 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<op::Scan>()) {
+      continue;
+    }
+    auto& scan = node.Cast<op::Scan>();
+    auto* source = std::get_if<op::Scan::Dataframe>(&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<std::string>& 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<core::StorageType>& 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<LogicalPlan> 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 0000000000..28db071096
--- /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 <cstdint>
+#include <string>
+#include <string_view>
+
+#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<LogicalPlan> 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 0000000000..564ecb4506
--- /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 <cstdint>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#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<PerfettoSqlParser::Pipeline>(parser.TakeStatement()).plan);
+  }
+
+  base::StatusOr<LogicalPlan> RoundTrip(const LogicalPlan& plan) {
+    return DeserializePlan(SerializePlan(plan), catalog_);
+  }
+
+  StringPool pool_;
+  std::unique_ptr<SqliteConnection> connection_;
+  TestCatalog catalog_;
+  LowerEnvironment env_;
+  base::FlatHashMap<std::string, PerfettoSqlParser::Macro> 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<std::string, size_t> 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<op::TreeAccumulate>().node_column =
+      reads_nothing_below.nodes[1]
+          .Cast<op::TreeAccumulate>()
+          .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 ba41d5e9ff..c05f4a585e 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 ca84de25fb..64b59536b8 100644
--- a/src/trace_processor/sqlite/sqlite_connection.h
+++ b/src/trace_processor/sqlite/sqlite_connection.h
@@ -20,7 +20,6 @@
 #include <sqlite3.h>
 #include <cstddef>
 #include <cstdint>
-#include <functional>
 #include <memory>
 #include <optional>
 #include <string>
@@ -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<void()> 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<void()> on_finalized;
-    };
-
     explicit PreparedStatement(ScopedStmt, SqlSource);
 
-    std::unique_ptr<sqlite3_stmt, Finalizer> stmt_;
+    ScopedStmt stmt_;
     ScopedSqliteString expanded_sql_;
     SqlSource sql_source_;
     base::Status status_ = base::OkStatus();