From ee626da4bf19880c6ac6f841143ed7f10f5e9f04 Mon Sep 17 00:00:00 2001 From: Lalit Maganti Date: Sun, 27 Sep 2026 20:12:23 +0100 Subject: [PATCH] tp: let pipelines read SQL through their arguments A pipeline ran its SQL sources itself, as statements of their own. So they could see nothing of the statement the pipeline was written in: not its CTEs, not the arguments of the function it was in, not the columns of an outer query. And the pipeline paid SQLite's row-at-a-time boundary twice, stepping each statement from outside. Now SQLite evaluates each relation a pipeline reads where the pipeline is written, and hands it over. The SQL a pipeline is replaced with collects each SQL source with an aggregate, `__intrinsic_rows`, into batches, and passes them to `__intrinsic_pipeline` after the plan: __intrinsic_pipeline(X'', (SELECT __intrinsic_rows('', a, b) FROM src)) The plan reads them as inputs by index, and never holds SQL: a SQL source is moved out into an input when the plan is written. Only the columns the pipeline uses are collected. Collecting is also cheaper than stepping: about 95 ms against 220 ms for 1M rows of three columns. A SQL source's columns now come from semantic analysis, never from preparing it, since only where the pipeline is written is everything it reads in scope: - dataframes and views, as before; - other tables SQLite knows, such as plain tables and table functions, by their schema (pragma_table_xinfo), untyped; - pipelines of the same statement, by the plan they compiled to, typed. A relation the analyzer cannot describe fails clearly. Leaf relations now own their strings, and the analyzer names columns as SQLite sees them, after macro expansion. Anyone can write a plan and its inputs into SQL, so a run checks that each input has the columns the plan reads from it. SqlScan, DescribeQuery and positional pruning of SQL sources are gone. Function arguments work with no special handling, as SQLite evaluates them in the collecting subquery. Needs syntaqlite 0.12, which expands pipelines once their statement is parsed. --- Android.bp | 6 +- BUILD | 8 +- src/perfetto_sql/analysis/relation.cc | 56 +- src/perfetto_sql/analysis/relation.h | 12 +- .../perfetto_sql/engine/connection_catalog.cc | 17 +- .../perfetto_sql/engine/connection_catalog.h | 5 +- .../engine/perfetto_sql_connection.cc | 15 +- .../engine/perfetto_sql_connection.h | 6 +- .../perfetto_sql_connection_unittest.cc | 42 +- .../perfetto_sql/engine/pipeline_module.cc | 133 ++++- .../perfetto_sql/engine/pipeline_module.h | 24 + .../perfetto_sql/exec/BUILD.gn | 7 +- .../perfetto_sql/exec/collected_rows.cc | 238 ++++++++ .../perfetto_sql/exec/collected_rows.h | 110 ++++ .../exec/collected_rows_unittest.cc | 152 +++++ .../perfetto_sql/exec/sql_scan.cc | 283 ---------- .../perfetto_sql/exec/sql_scan.h | 102 ---- .../perfetto_sql/exec/sql_scan_unittest.cc | 532 ------------------ .../perfetto_sql/parser/BUILD.gn | 2 + .../parser/perfetto_sql_parser.cc | 69 ++- .../parser/perfetto_sql_parser_unittest.cc | 55 +- .../perfetto_sql/pipeline/BUILD.gn | 2 + .../perfetto_sql/pipeline/catalog.h | 17 +- .../perfetto_sql/pipeline/column_pruning.cc | 37 +- .../perfetto_sql/pipeline/compiler.cc | 70 ++- .../perfetto_sql/pipeline/logical_plan.h | 13 +- .../pipeline/logical_plan_test_utils.cc | 5 + .../perfetto_sql/pipeline/physical_plan.cc | 59 +- .../perfetto_sql/pipeline/physical_plan.h | 17 +- .../pipeline/physical_plan_unittest.cc | 26 +- .../perfetto_sql/pipeline/pipeline_sql.cc | 131 ++++- .../perfetto_sql/pipeline/pipeline_sql.h | 25 +- .../pipeline/plan_serialization.cc | 27 +- .../pipeline/plan_serialization_unittest.cc | 15 +- .../perfetto_sql/pipeline/test_catalog.h | 74 ++- .../perfetto_sql/schema/BUILD.gn | 4 +- .../perfetto_sql/schema/query_schema.cc | 112 ---- .../perfetto_sql/schema/sqlite_relations.cc | 53 ++ .../{query_schema.h => sqlite_relations.h} | 25 +- 39 files changed, 1307 insertions(+), 1279 deletions(-) create mode 100644 src/trace_processor/perfetto_sql/exec/collected_rows.cc create mode 100644 src/trace_processor/perfetto_sql/exec/collected_rows.h create mode 100644 src/trace_processor/perfetto_sql/exec/collected_rows_unittest.cc delete mode 100644 src/trace_processor/perfetto_sql/exec/sql_scan.cc delete mode 100644 src/trace_processor/perfetto_sql/exec/sql_scan.h delete mode 100644 src/trace_processor/perfetto_sql/exec/sql_scan_unittest.cc delete mode 100644 src/trace_processor/perfetto_sql/schema/query_schema.cc create mode 100644 src/trace_processor/perfetto_sql/schema/sqlite_relations.cc rename src/trace_processor/perfetto_sql/schema/{query_schema.h => sqlite_relations.h} (55%) diff --git a/Android.bp b/Android.bp index 705dd538123..a967f781db2 100644 --- a/Android.bp +++ b/Android.bp @@ -19005,7 +19005,7 @@ filegroup { filegroup { name: "perfetto_src_trace_processor_perfetto_sql_exec_exec", srcs: [ - "src/trace_processor/perfetto_sql/exec/sql_scan.cc", + "src/trace_processor/perfetto_sql/exec/collected_rows.cc", ], } @@ -19013,7 +19013,7 @@ filegroup { filegroup { name: "perfetto_src_trace_processor_perfetto_sql_exec_unittests", srcs: [ - "src/trace_processor/perfetto_sql/exec/sql_scan_unittest.cc", + "src/trace_processor/perfetto_sql/exec/collected_rows_unittest.cc", ], } @@ -19121,7 +19121,7 @@ filegroup { filegroup { name: "perfetto_src_trace_processor_perfetto_sql_schema_schema", srcs: [ - "src/trace_processor/perfetto_sql/schema/query_schema.cc", + "src/trace_processor/perfetto_sql/schema/sqlite_relations.cc", ], } diff --git a/BUILD b/BUILD index fa2209888f8..b21b48dfa4d 100644 --- a/BUILD +++ b/BUILD @@ -3829,8 +3829,8 @@ perfetto_filegroup( perfetto_filegroup( name = "src_trace_processor_perfetto_sql_exec_exec", srcs = [ - "src/trace_processor/perfetto_sql/exec/sql_scan.cc", - "src/trace_processor/perfetto_sql/exec/sql_scan.h", + "src/trace_processor/perfetto_sql/exec/collected_rows.cc", + "src/trace_processor/perfetto_sql/exec/collected_rows.h", ], ) @@ -3901,8 +3901,8 @@ perfetto_filegroup( perfetto_filegroup( name = "src_trace_processor_perfetto_sql_schema_schema", srcs = [ - "src/trace_processor/perfetto_sql/schema/query_schema.cc", - "src/trace_processor/perfetto_sql/schema/query_schema.h", + "src/trace_processor/perfetto_sql/schema/sqlite_relations.cc", + "src/trace_processor/perfetto_sql/schema/sqlite_relations.h", "src/trace_processor/perfetto_sql/schema/type_mapping.h", ], ) diff --git a/src/perfetto_sql/analysis/relation.cc b/src/perfetto_sql/analysis/relation.cc index 29477fa9242..14f186a3cb6 100644 --- a/src/perfetto_sql/analysis/relation.cc +++ b/src/perfetto_sql/analysis/relation.cc @@ -51,8 +51,14 @@ struct OwnedView { uint32_t root = 0; }; +// The text SQLite sees for the name at `span`, after macro expansion. A name is +// one token, so its text is a slice of the one layer it is in, which lives as +// long as the statement. std::string_view Text(SyntaqliteParser* p, SyntaqliteTextSpan span) { - return base::TrimWhitespace(SyntaqliteSpanText(p, span)); + uint32_t len = 0; + const char* text = syntaqlite_parser_span_expanded_text(p, &span, &len); + return text ? base::TrimWhitespace(std::string_view(text, len)) + : std::string_view(); } const SyntaqliteNode* Node(SyntaqliteParser* p, uint32_t id) { @@ -150,6 +156,7 @@ class RelationAnalyzer::Impl { void Begin() { preserves_rows_ = true; views_.clear(); + leaves_.clear(); } // The columns of the relation `name`. Its hidden columns, if any, are added @@ -185,15 +192,38 @@ class RelationAnalyzer::Impl { static ColumnLineage Lookup(const Scope&, std::string_view table, std::string_view column); + // The columns of a leaf relation, kept for the rest of the analysis. Its + // hidden columns are added to `hidden`. + std::vector LeafColumns(LeafRelation, + std::vector& hidden); const Catalog& catalog_; // Lineage string_views point into each view's sql string and parse tree, so // every OwnedView needs a stable address: growing a std::vector // would move the elements and moving `sql` can relocate its bytes (SSO). std::vector> views_; + // Lineage string_views point into each leaf relation's strings, so they are + // kept at stable addresses for the same reason. + std::vector> leaves_; bool preserves_rows_ = true; }; +std::vector RelationAnalyzer::Impl::LeafColumns( + LeafRelation found, + std::vector& hidden) { + leaves_.push_back(std::make_unique(std::move(found))); + const LeafRelation* relation = leaves_.back().get(); + std::vector out; + out.reserve(relation->columns.size()); + for (const LeafColumn& column : relation->columns) { + out.push_back({column.name, {{relation->name, column.name, column.type}}}); + if (column.hidden) { + hidden.push_back(column.name); + } + } + return out; +} + ColumnLineage RelationAnalyzer::Impl::Lookup(const Scope& scope, std::string_view table, std::string_view column) { @@ -436,8 +466,15 @@ RelationAnalyzer::Impl::Select(SyntaqliteParser* p, uint32_t id, int depth) { } return std::move(*left); } - default: - return base::ErrStatus("relation analysis: not a select"); + default: { + std::optional found = catalog_.FindNodeRelation({p, id}); + if (!found) { + return base::ErrStatus("relation analysis: not a select"); + } + preserves_rows_ = false; + std::vector hidden; + return LeafColumns(std::move(*found), hidden); + } } } @@ -445,17 +482,8 @@ base::StatusOr> RelationAnalyzer::Impl::Relation( std::string_view name, int depth, std::vector& hidden) { - if (std::optional relation = catalog_.FindLeafRelation(name)) { - std::vector out; - out.reserve(relation->columns.size()); - for (const LeafColumn& column : relation->columns) { - out.push_back( - {column.name, {{relation->name, column.name, column.type}}}); - if (column.hidden) { - hidden.push_back(column.name); - } - } - return out; + if (std::optional found = catalog_.FindLeafRelation(name)) { + return LeafColumns(std::move(*found), hidden); } if (depth >= kMaxDepth) { return base::ErrStatus( diff --git a/src/perfetto_sql/analysis/relation.h b/src/perfetto_sql/analysis/relation.h index c832464a9ad..c0acbbf578e 100644 --- a/src/perfetto_sql/analysis/relation.h +++ b/src/perfetto_sql/analysis/relation.h @@ -50,7 +50,7 @@ using ColumnType = base::TypeSet; // A leaf relation whose columns can be used as lineage origins. struct LeafColumn { - std::string_view name; + std::string name; // Nothing when the catalog does not know how the column is stored. std::optional type; // Left out of `*` and `table.*`, as SQLite does for HIDDEN columns, but @@ -58,12 +58,11 @@ struct LeafColumn { bool hidden = false; }; struct LeafRelation { - std::string_view name; + std::string name; std::vector columns; }; -// Supplies the schema objects referenced by parsed queries. Returned leaf -// strings only need to remain valid for the duration of an Analyze call. +// Supplies the schema objects referenced by parsed queries. class Catalog { public: virtual ~Catalog(); @@ -72,6 +71,11 @@ class Catalog { std::string_view name) const = 0; virtual std::optional FindViewSql( std::string_view name) const = 0; + // The relation a node of the dialect's own stands for, such as a pipeline + // written as a subquery, when the host knows it. + virtual std::optional FindNodeRelation(SqlNode) const { + return std::nullopt; + } }; struct ColumnOrigin { diff --git a/src/trace_processor/perfetto_sql/engine/connection_catalog.cc b/src/trace_processor/perfetto_sql/engine/connection_catalog.cc index 8cf1357d0d5..7608b71d066 100644 --- a/src/trace_processor/perfetto_sql/engine/connection_catalog.cc +++ b/src/trace_processor/perfetto_sql/engine/connection_catalog.cc @@ -28,7 +28,7 @@ #include "src/trace_processor/core/dataframe/dataframe.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/schema/query_schema.h" +#include "src/trace_processor/perfetto_sql/schema/sqlite_relations.h" #include "src/trace_processor/perfetto_sql/schema/type_mapping.h" #include "src/trace_processor/sqlite/bindings/sqlite_column.h" #include "src/trace_processor/sqlite/sql_source.h" @@ -77,10 +77,15 @@ std::optional ConnectionCatalog::FindLeafRelation( std::string_view name) const { const dataframe::Dataframe* dataframe = connection_->GetDataframeOrNull(name); if (!dataframe) { - return std::nullopt; + // A view is expanded instead, so its columns are traced through it. + if (FindViewSql(name)) { + return std::nullopt; + } + return sql_schema::FindSqliteRelation(connection_->sqlite_connection(), + name); } analysis::LeafRelation relation; - relation.name = name; + relation.name = std::string(name); const std::vector& columns = dataframe->column_names(); relation.columns.reserve(columns.size()); for (uint32_t i = 0; i < columns.size(); ++i) { @@ -109,10 +114,4 @@ const dataframe::Dataframe* ConnectionCatalog::FindDataframe( return connection_->GetDataframeOrNull(name); } -base::StatusOr ConnectionCatalog::DescribeQuery( - const SqlSource& sql) const { - return sql_schema::DescribeQuery(connection_->sqlite_connection(), sql, - *this); -} - } // namespace perfetto::trace_processor diff --git a/src/trace_processor/perfetto_sql/engine/connection_catalog.h b/src/trace_processor/perfetto_sql/engine/connection_catalog.h index 7221007cf1a..f573e034932 100644 --- a/src/trace_processor/perfetto_sql/engine/connection_catalog.h +++ b/src/trace_processor/perfetto_sql/engine/connection_catalog.h @@ -33,8 +33,7 @@ namespace perfetto::trace_processor { // Adapts a connection to semantic analysis and pipeline compilation. Each // dataframe is served as a typed leaf relation and as a dataframe. -class ConnectionCatalog final : public perfetto_sql::analysis::Catalog, - public pipeline::Catalog { +class ConnectionCatalog final : public pipeline::Catalog { public: explicit ConnectionCatalog(PerfettoSqlConnection*); @@ -44,8 +43,6 @@ class ConnectionCatalog final : public perfetto_sql::analysis::Catalog, const dataframe::Dataframe* FindDataframe( std::string_view name) const override; - base::StatusOr DescribeQuery( - const SqlSource& sql) const override; private: PerfettoSqlConnection* connection_; 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 cfa7433413a..079a742c6a9 100644 --- a/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.cc +++ b/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.cc @@ -364,6 +364,8 @@ PerfettoSqlConnection::PerfettoSqlConnection( ctx->connection = this; RegisterVirtualTableModule(pipeline::kPipelineFunction, std::move(ctx)); + base::Status status = RegisterAggregateFunction(pool_); + PERFETTO_CHECK(status.ok()); } database_->InitializeSharedSchema(connection_.get()); @@ -1057,18 +1059,23 @@ PerfettoSqlConnection::PreparePipeline(const pipeline::LogicalPlan& plan, return base::ErrStatus("%s%s", source.AsTraceback(0).c_str(), sql.status().c_message()); } - return connection_->PrepareStatement(source.RewriteAllIgnoreExisting( - SqlSource::FromTraceProcessorImplementation(std::move(*sql)))); + SqliteConnection::PreparedStatement stmt = + connection_->PrepareStatement(source.RewriteAllIgnoreExisting( + SqlSource::FromTraceProcessorImplementation(std::move(*sql)))); + RETURN_IF_ERROR(stmt.status()); + return std::move(stmt); } base::StatusOr> -PerfettoSqlConnection::LoadPipeline(std::string_view serialized) { +PerfettoSqlConnection::LoadPipeline( + std::string_view serialized, + const exec::CollectedRowsScan::Inputs& inputs) { 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_; + env.inputs = &inputs; return pipeline::Lower(plan, env); } 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 895222dc9b8..a682fdf3d7b 100644 --- a/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.h +++ b/src/trace_processor/perfetto_sql/engine/perfetto_sql_connection.h @@ -174,9 +174,11 @@ class PerfettoSqlConnection { base::StatusOr PrepareSqliteStatement( SqlSource sql); - // Loads a plan written by pipeline::SerializePlan, ready to run. + // Loads a plan written by pipeline::SerializePlan, ready to run. Each run + // reads the rows of the plan's inputs from `inputs`, which must outlive it. base::StatusOr> LoadPipeline( - std::string_view serialized); + std::string_view serialized, + const exec::CollectedRowsScan::Inputs& inputs); // Registers a virtual table module with the given name. // 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 696374b9101..f61530f0007 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 @@ -856,7 +856,9 @@ class PerfettoSqlConnectionPipelineTest : public PerfettoSqlConnectionTest { std::string PipelineSql(const std::string& pipeline) { auto res = connection_->ExecuteUntilLastStatement( SqlSource::FromExecuteQuery(pipeline)); - PERFETTO_CHECK(res.ok()); + if (!res.ok()) { + PERFETTO_FATAL("%s", res.status().c_message()); + } return res->stmt.sql(); } @@ -1176,17 +1178,25 @@ TEST_F(PerfettoSqlConnectionPipelineTest, PipelinesInViewsAndFunctions) { EXPECT_THAT(*rows, testing::ElementsAre("20")); } -// A pipeline runs its SQL sources itself, so nothing binds a function's -// arguments there: reading one is refused when the pipeline is compiled, not -// silently NULL. -TEST_F(PerfettoSqlConnectionPipelineTest, PipelineSourcesCannotReadArguments) { - EXPECT_THAT( - Rows("CREATE PERFETTO FUNCTION scaled(k LONG) RETURNS TABLE(total LONG) " - "AS SELECT total FROM (FROM (SELECT id, parent_id, self * $k AS " +// A pipeline's SQL is evaluated where the pipeline is written, so it reads +// the arguments of the function it is in like any other SQL there. +TEST_F(PerfettoSqlConnectionPipelineTest, PipelineSourcesReadArguments) { + ASSERT_TRUE( + Rows("CREATE PERFETTO FUNCTION scaled(k LONG) " + "RETURNS TABLE(id LONG, total LONG) " + "AS SELECT id, total FROM (FROM (SELECT id, parent_id, self * $k AS " "self FROM tree) |> TREE ACCUMULATE UP SUM(self) AS total)") - .status() - .message(), - testing::HasSubstr("Cannot read `$k`")); + .ok()); + auto rows = Rows("SELECT id, total FROM scaled(2)"); + ASSERT_TRUE(rows.ok()) << rows.status().message(); + EXPECT_THAT(*rows, testing::ElementsAre("0,200", "1,120", "2,60", "3,80")); + + // Read again for each row of a join, with a different argument each time. + rows = Rows( + "SELECT k.v, s.total FROM (SELECT 1 AS v UNION ALL SELECT 2 " + "UNION ALL SELECT 1) k JOIN scaled(k.v) s WHERE s.id = 0"); + ASSERT_TRUE(rows.ok()) << rows.status().message(); + EXPECT_THAT(*rows, testing::UnorderedElementsAre("1,100", "2,200", "1,100")); } // Macros and pipelines nest inside each other in every way: pipelines in @@ -1239,9 +1249,9 @@ TEST_F(PerfettoSqlConnectionPipelineTest, PipelinesAsIntersectionOperands) { EXPECT_THAT(*rows, testing::ElementsAre("12,3", "20,2")); } -// SQLite reads the inner side of a join once per outer row. A pipeline read -// again with the same plan is not run again each time: `random()` in its -// source would otherwise give each read different rows. +// SQLite reads the inner side of a join once per outer row. The relation a +// pipeline reads is collected once, so every read sees the same rows even when +// its SQL, like `random()`, would give different ones each time. TEST_F(PerfettoSqlConnectionPipelineTest, PipelinesReadAgainAreNotRunAgain) { ASSERT_TRUE(Rows("CREATE TABLE picks(k); " "INSERT INTO picks VALUES (1), (2), (3), (4), (5)") @@ -1250,9 +1260,7 @@ TEST_F(PerfettoSqlConnectionPipelineTest, PipelinesReadAgainAreNotRunAgain) { "SELECT count(DISTINCT p.r) FROM picks CROSS JOIN " "(FROM (SELECT 1 AS one, random() AS r)) p"); ASSERT_TRUE(rows.ok()) << rows.status().message(); - // The second read runs the pipeline once more to keep its rows; every read - // after replays them. - EXPECT_THAT(*rows, testing::ElementsAre("2")); + EXPECT_THAT(*rows, testing::ElementsAre("1")); } TEST_F(PerfettoSqlConnectionPipelineTest, PipelineSubqueriesNeedPipelines) { diff --git a/src/trace_processor/perfetto_sql/engine/pipeline_module.cc b/src/trace_processor/perfetto_sql/engine/pipeline_module.cc index 46d709da8f4..825874f93c6 100644 --- a/src/trace_processor/perfetto_sql/engine/pipeline_module.cc +++ b/src/trace_processor/perfetto_sql/engine/pipeline_module.cc @@ -18,6 +18,7 @@ #include +#include #include #include #include @@ -51,14 +52,24 @@ namespace { using core::exec::ColumnView; using core::exec::Variant; -// The text 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. +// The table function's arguments: the plan, then the rows of its inputs. They +// come 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 kTextColumn = 0; -constexpr int kFirstOutputColumn = 1; +constexpr int kFirstInputColumn = 1; +constexpr int kFirstOutputColumn = + kFirstInputColumn + static_cast(pipeline::kMaxPipelineInputs); + +// idxNum: whether an output rowid is looked up, and how many inputs are given. +constexpr int kRowidLookup = 1; +constexpr int kInputCountShift = 1; std::string Schema() { // Public names (which may repeat) are applied by the outer SELECT. std::vector columns{"pipeline HIDDEN"}; + for (uint32_t i = 0; i < pipeline::kMaxPipelineInputs; ++i) { + columns.push_back("i" + std::to_string(i) + " HIDDEN"); + } for (uint32_t i = 0; i < pipeline::kMaxPipelineColumns; ++i) { columns.push_back("c" + std::to_string(i)); } @@ -168,6 +179,36 @@ void ResetPlan(PipelineModule::Cursor* c) { c->serialized.clear(); } +// Takes the rows SQLite collected for each input, or fails on an argument +// which is not a collection of what the plan reads. +int TakeInputs(PipelineModule::Cursor* c, + sqlite3_vtab* vtab, + sqlite3_value** values, + int count) { + exec::CollectedRowsScan::Inputs inputs; + for (int i = 0; i < count; ++i) { + const auto* rows = + static_cast*>( + sqlite3_value_pointer(values[i], exec::kCollectedRowsPointerType)); + if (!rows) { + return sqlite::utils::SetError( + vtab, "__intrinsic_pipeline: expected collected rows"); + } + inputs.push_back(*rows); + } + if (base::Status status = c->plan->CheckInputs(inputs); !status.ok()) { + return sqlite::utils::SetError( + vtab, base::ErrStatus("__intrinsic_pipeline: %s", status.c_message())); + } + // Different rows give different output, so nothing kept can be replayed. + if (inputs != c->inputs) { + c->rows.reset(); + c->memoize.reset(); + c->inputs = std::move(inputs); + } + return SQLITE_OK; +} + // Surfaces the executor's error, if any, once rows stop. int CheckStatus(PipelineModule::Cursor* cursor) { // status() walks every node so only check it once rows stop. @@ -178,6 +219,50 @@ int CheckStatus(PipelineModule::Cursor* cursor) { } // namespace +void CollectRows::Step(sqlite3_context* ctx, int argc, sqlite3_value** argv) { + AggCtx& agg = AggCtx::GetOrCreateContextForStep(ctx); + if (!agg.status.ok()) { + return; + } + if (!agg.rows) { + const auto* text = + reinterpret_cast(sqlite3_value_text(argv[0])); + base::StatusOr columns = pipeline::ReadCollectedColumns( + text ? std::string_view(text) : std::string_view()); + if (columns.ok() && columns->size() != static_cast(argc - 1)) { + columns = base::ErrStatus("%s: expected %zu values, not %d", kName, + columns->size(), argc - 1); + } + if (!columns.ok()) { + agg.status = columns.status(); + return sqlite::result::Error(ctx, agg.status.c_message()); + } + agg.rows = std::make_shared(std::move(*columns), + GetUserData(ctx)); + } + agg.status = agg.rows->Append(argv + 1); + if (!agg.status.ok()) { + return sqlite::result::Error(ctx, agg.status.c_message()); + } +} + +void CollectRows::Final(sqlite3_context* ctx) { + auto agg = AggCtx::GetContextOrNullForFinal(ctx); + std::shared_ptr rows; + if (agg.get() && agg.get()->rows) { + rows = std::move(agg.get()->rows); + } else { + // No rows, so nothing needs their columns. + rows = + std::make_shared(core::Schema(), GetUserData(ctx)); + } + return sqlite::result::UniquePointer( + ctx, + std::make_unique>( + std::move(rows)), + exec::kCollectedRowsPointerType); +} + int PipelineModule::Connect(sqlite3* db, void* raw_ctx, int, @@ -201,16 +286,28 @@ int PipelineModule::Disconnect(sqlite3_vtab* vtab) { int PipelineModule::BestIndex(sqlite3_vtab*, sqlite3_index_info* info) { int text = -1; int rowid = -1; + // The constraint giving each input's rows, by position. + std::vector inputs(pipeline::kMaxPipelineInputs, -1); + int input_count = 0; for (int i = 0; i < info->nConstraint; ++i) { const auto& constraint = info->aConstraint[i]; if (constraint.op != SQLITE_INDEX_CONSTRAINT_EQ) continue; + // Without its plan and inputs there is nothing to run, so no plan which + // lacks them can be usable. if (constraint.iColumn == kTextColumn) { - // Without its text there is nothing to run, so no plan can be usable. if (!constraint.usable) return SQLITE_CONSTRAINT; text = i; } + if (constraint.iColumn >= kFirstInputColumn && + constraint.iColumn < kFirstOutputColumn) { + if (!constraint.usable) + return SQLITE_CONSTRAINT; + int position = constraint.iColumn - kFirstInputColumn; + inputs[static_cast(position)] = i; + input_count = std::max(input_count, position + 1); + } if (constraint.iColumn == -1 && constraint.usable) rowid = i; } @@ -218,10 +315,19 @@ int PipelineModule::BestIndex(sqlite3_vtab*, sqlite3_index_info* info) { return SQLITE_CONSTRAINT; info->aConstraintUsage[text].argvIndex = 1; info->aConstraintUsage[text].omit = true; + for (int position = 0; position < input_count; ++position) { + int i = inputs[static_cast(position)]; + // Arguments are given in order, so none can be missing before the last. + if (i == -1) + return SQLITE_CONSTRAINT; + info->aConstraintUsage[i].argvIndex = 2 + position; + info->aConstraintUsage[i].omit = true; + } + info->idxNum = input_count << kInputCountShift; if (rowid != -1) { - info->aConstraintUsage[rowid].argvIndex = 2; + info->aConstraintUsage[rowid].argvIndex = 2 + input_count; // SQLite rechecks comparisons, including non-integer RHS values. - info->idxNum = 1; + info->idxNum |= kRowidLookup; info->estimatedRows = 1; } // Output rowid constraints run after the fold. General predicates remain @@ -246,7 +352,9 @@ int PipelineModule::Filter(sqlite3_vtab_cursor* cursor, int argc, sqlite3_value** argv) { Cursor* c = GetCursor(cursor); - PERFETTO_DCHECK(argc == (idx_num ? 2 : 1)); + int input_count = idx_num >> kInputCountShift; + bool rowid_lookup = idx_num & kRowidLookup; + PERFETTO_DCHECK(argc == 1 + input_count + (rowid_lookup ? 1 : 0)); if (sqlite3_value_type(argv[0]) != SQLITE_BLOB) { return sqlite::utils::SetError(cursor->pVtab, "__intrinsic_pipeline: expected a plan"); @@ -256,7 +364,7 @@ int PipelineModule::Filter(sqlite3_vtab_cursor* cursor, static_cast(sqlite3_value_bytes(argv[0]))); if (!c->plan || c->serialized != serialized) { Context* context = GetVtab(cursor->pVtab)->context; - auto plan = context->connection->LoadPipeline(serialized); + auto plan = context->connection->LoadPipeline(serialized, c->inputs); if (!plan.ok()) { return sqlite::utils::SetError(cursor->pVtab, plan.status()); } @@ -265,6 +373,10 @@ int PipelineModule::Filter(sqlite3_vtab_cursor* cursor, c->serialized = serialized; c->pool = context->pool; } + if (int rc = TakeInputs(c, cursor->pVtab, argv + 1, input_count); + rc != SQLITE_OK) { + return rc; + } // SQLite reads the inner side of a join again for each outer row. The // second read of a plan keeps what it produces, and later reads replay it. if (!c->rows) { @@ -276,8 +388,9 @@ int PipelineModule::Filter(sqlite3_vtab_cursor* cursor, } 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 (rowid_lookup && + sqlite3_value_type(argv[1 + input_count]) == SQLITE_INTEGER) { + c->target_rowid = sqlite3_value_int64(argv[1 + input_count]); if (*c->target_rowid < 0) { c->eof = true; return SQLITE_OK; diff --git a/src/trace_processor/perfetto_sql/engine/pipeline_module.h b/src/trace_processor/perfetto_sql/engine/pipeline_module.h index 0caaead96a5..306013a0fae 100644 --- a/src/trace_processor/perfetto_sql/engine/pipeline_module.h +++ b/src/trace_processor/perfetto_sql/engine/pipeline_module.h @@ -29,8 +29,11 @@ #include "src/trace_processor/containers/string_pool.h" #include "src/trace_processor/core/exec/memoize.h" #include "src/trace_processor/core/exec/row_cursor.h" +#include "src/trace_processor/perfetto_sql/exec/collected_rows.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/pipeline_sql.h" +#include "src/trace_processor/sqlite/bindings/sqlite_aggregate_function.h" #include "src/trace_processor/sqlite/bindings/sqlite_module.h" namespace perfetto::trace_processor { @@ -47,6 +50,24 @@ class PerfettoSqlConnection; // // The table has a fixed width: pipelines output into generic columns `c0`, // `c1`, ..., which the SQL reading them renames to the pipeline's own names. +// `__intrinsic_rows(columns, value, ...)`: collects the rows of a relation for +// a pipeline to read, as a pointer to a std::shared_ptr +// under exec::kCollectedRowsPointerType. `columns` says what each value is +// (see pipeline::WriteCollectedColumns). +struct CollectRows : sqlite::AggregateFunction { + static constexpr const char* kName = pipeline::kCollectFunction; + static constexpr int kArgCount = -1; + using UserData = StringPool; + + struct AggCtx : sqlite::AggregateContext { + std::shared_ptr rows; + base::Status status; + }; + + static void Step(sqlite3_context*, int argc, sqlite3_value** argv); + static void Final(sqlite3_context*); +}; + struct PipelineModule : sqlite::Module { static constexpr auto kType = kEponymousOnly; static constexpr bool kSupportsWrites = false; @@ -71,6 +92,9 @@ struct PipelineModule : sqlite::Module { }; // The serialized plan `plan` was loaded from. std::string serialized; + // The rows of the plan's inputs, which it reads on each run. Declared + // before the plan, which reads them. + exec::CollectedRowsScan::Inputs inputs; std::unique_ptr plan; // Keeps what `plan` produced once the cursor is read again. std::unique_ptr memoize; diff --git a/src/trace_processor/perfetto_sql/exec/BUILD.gn b/src/trace_processor/perfetto_sql/exec/BUILD.gn index d559857494d..272b91ea0ac 100644 --- a/src/trace_processor/perfetto_sql/exec/BUILD.gn +++ b/src/trace_processor/perfetto_sql/exec/BUILD.gn @@ -16,8 +16,8 @@ import("../../../../gn/test.gni") source_set("exec") { sources = [ - "sql_scan.cc", - "sql_scan.h", + "collected_rows.cc", + "collected_rows.h", ] deps = [ "../../../../gn:default_deps", @@ -33,7 +33,7 @@ source_set("exec") { perfetto_unittest_source_set("unittests") { testonly = true - sources = [ "sql_scan_unittest.cc" ] + sources = [ "collected_rows_unittest.cc" ] deps = [ ":exec", "../../../../gn:default_deps", @@ -46,6 +46,5 @@ perfetto_unittest_source_set("unittests") { "../../core/exec:test_utils", "../../core/util", "../../sqlite", - "../schema", ] } diff --git a/src/trace_processor/perfetto_sql/exec/collected_rows.cc b/src/trace_processor/perfetto_sql/exec/collected_rows.cc new file mode 100644 index 00000000000..7506cda5174 --- /dev/null +++ b/src/trace_processor/perfetto_sql/exec/collected_rows.cc @@ -0,0 +1,238 @@ +/* + * 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/exec/collected_rows.h" + +#include + +#include +#include +#include +#include +#include +#include + +#include "perfetto/base/compiler.h" +#include "perfetto/base/logging.h" +#include "perfetto/base/status.h" +#include "src/trace_processor/containers/string_pool.h" +#include "src/trace_processor/core/common/storage_types.h" +#include "src/trace_processor/core/exec/column_chunk.h" +#include "src/trace_processor/core/exec/column_view.h" +#include "src/trace_processor/core/exec/operator.h" +#include "src/trace_processor/core/exec/row_batch.h" +#include "src/trace_processor/core/exec/row_selection.h" +#include "src/trace_processor/core/exec/variant.h" + +namespace perfetto::trace_processor::exec { +namespace { + +using core::StorageType; +using core::exec::ColumnChunk; +using core::exec::ColumnView; +using core::exec::kMaxBatchRows; +using core::exec::RowBatch; +using core::exec::RowSelection; +using core::exec::Variant; + +StringPool::Id InternText(StringPool* pool, sqlite3_value* value) { + const auto* text = reinterpret_cast(sqlite3_value_text(value)); + return pool->InternString( + std::string_view(text, static_cast(sqlite3_value_bytes(value)))); +} + +} // namespace + +CollectedRows::CollectedRows(core::Schema columns, StringPool* pool) + : columns_(std::move(columns)), pool_(pool) {} + +CollectedRows::~CollectedRows() = default; + +void CollectedRows::AddBatch() { + Batch batch; + for (const core::ColumnSchema& column : columns_) { + auto chunk = std::make_shared(); + void* data = nullptr; + if (!column.type) { + data = chunk->Values().data(); + } else { + switch (column.type->index()) { + case StorageType::GetTypeIndex(): + data = chunk->Values().data(); + break; + case StorageType::GetTypeIndex(): + data = chunk->Values().data(); + break; + case StorageType::GetTypeIndex(): + data = chunk->Values().data(); + break; + case StorageType::GetTypeIndex(): + data = chunk->Values().data(); + break; + case StorageType::GetTypeIndex(): + data = chunk->Values().data(); + break; + default: + // An Id has no storage of its own, so it is collected as a Uint32. + PERFETTO_FATAL("Unreachable"); + } + chunk->validity.resize(kMaxBatchRows); + } + batch.columns.push_back(std::move(chunk)); + batch.data.push_back(data); + } + batches_.push_back(std::move(batch)); +} + +base::Status CollectedRows::Append(sqlite3_value** values) { + if (batches_.empty() || batches_.back().count == kMaxBatchRows) { + AddBatch(); + } + Batch& batch = batches_.back(); + for (uint32_t i = 0; i < columns_.size(); ++i) { + base::Status status = Read(batch, i, values[i]); + if (PERFETTO_UNLIKELY(!status.ok())) { + return status; + } + } + ++batch.count; + return base::OkStatus(); +} + +base::Status CollectedRows::Read(Batch& batch, + uint32_t column, + sqlite3_value* value) { + if (!columns_[column].type) { + auto* data = static_cast(batch.data[column]); + Variant& out = data[batch.count]; + switch (sqlite3_value_type(value)) { + case SQLITE_INTEGER: + out = Variant::Int64(sqlite3_value_int64(value)); + return base::OkStatus(); + case SQLITE_FLOAT: + out = Variant::Double(sqlite3_value_double(value)); + return base::OkStatus(); + case SQLITE_TEXT: + out = Variant::String(InternText(pool_, value)); + return base::OkStatus(); + case SQLITE_NULL: + out = Variant::Null(); + return base::OkStatus(); + default: + return base::ErrStatus( + "column '%s' holds a blob, which a pipeline cannot carry", + columns_[column].name.c_str()); + } + } + switch (columns_[column].type->index()) { + case StorageType::GetTypeIndex(): + return ReadTyped(batch, column, value); + case StorageType::GetTypeIndex(): + return ReadTyped(batch, column, value); + case StorageType::GetTypeIndex(): + return ReadTyped(batch, column, value); + case StorageType::GetTypeIndex(): + return ReadTyped(batch, column, value); + case StorageType::GetTypeIndex(): + return ReadTyped(batch, column, value); + default: + PERFETTO_FATAL("Unreachable"); + } +} + +template +base::Status CollectedRows::ReadTyped(Batch& batch, + uint32_t column, + sqlite3_value* value) { + auto* data = static_cast(batch.data[column]); + uint32_t row = batch.count; + int type = sqlite3_value_type(value); + if (PERFETTO_LIKELY(type == kSqliteType)) { + if constexpr (std::is_same_v) { + data[row] = InternText(pool_, value); + } else if constexpr (std::is_same_v) { + data[row] = sqlite3_value_double(value); + } else { + data[row] = static_cast(sqlite3_value_int64(value)); + } + batch.columns[column]->validity.set(row); + return base::OkStatus(); + } + if (type == SQLITE_NULL) { + // The row is null, but write the slot anyway. A flat column's storage is + // readable at every row, so a reader summing it needs no per-row branch. + if constexpr (std::is_same_v) { + data[row] = StringPool::Id::Null(); + } else { + data[row] = T{}; + } + return base::OkStatus(); + } + // Only reachable if the type the column was traced back to was wrong. + return base::ErrStatus("column '%s' does not hold what it was traced back to", + columns_[column].name.c_str()); +} + +void CollectedRows::View(uint32_t i, RowBatch& out) const { + const Batch& batch = batches_[i]; + out.Reset(); + for (uint32_t c = 0; c < columns_.size(); ++c) { + const std::shared_ptr& chunk = batch.columns[c]; + if (!columns_[c].type) { + out.AddColumn( + ColumnView::Variants(static_cast(batch.data[c])), + chunk); + } else { + out.AddColumn(ColumnView::Reference(*columns_[c].type, batch.data[c], + &chunk->validity), + chunk); + } + } + out.Compose(RowSelection::Range(0), batch.count); + out.SetCardinality(batch.count); +} + +CollectedRowsScan::CollectedRowsScan(const Inputs& inputs, uint32_t index) + : inputs_(inputs), index_(index) {} + +CollectedRowsScan::~CollectedRowsScan() = default; +CollectedRowsScan::State::~State() = default; + +std::unique_ptr CollectedRowsScan::MakeState() + const { + return std::make_unique(); +} + +bool CollectedRowsScan::GetData(RowBatch& out, + core::exec::OperatorState& state) const { + State& s = state.Cast(); + const CollectedRows& rows = *inputs_[index_]; + if (s.next_batch == rows.batch_count()) { + return false; + } + rows.View(s.next_batch++, out); + return true; +} + +void CollectedRowsScan::Rewind(core::exec::OperatorState& state) const { + state.Cast().next_batch = 0; +} + +base::Status CollectedRowsScan::status(const core::exec::OperatorState&) const { + return base::OkStatus(); +} + +} // namespace perfetto::trace_processor::exec diff --git a/src/trace_processor/perfetto_sql/exec/collected_rows.h b/src/trace_processor/perfetto_sql/exec/collected_rows.h new file mode 100644 index 00000000000..8007c417ed5 --- /dev/null +++ b/src/trace_processor/perfetto_sql/exec/collected_rows.h @@ -0,0 +1,110 @@ +/* + * 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_EXEC_COLLECTED_ROWS_H_ +#define SRC_TRACE_PROCESSOR_PERFETTO_SQL_EXEC_COLLECTED_ROWS_H_ + +#include +#include +#include + +#include "perfetto/base/status.h" +#include "src/trace_processor/containers/string_pool.h" +#include "src/trace_processor/core/common/schema.h" +#include "src/trace_processor/core/exec/column_chunk.h" +#include "src/trace_processor/core/exec/operator.h" +#include "src/trace_processor/core/exec/row_batch.h" + +struct sqlite3_value; + +namespace perfetto::trace_processor::exec { + +// The pointer type SQLite passes collected rows under: a +// std::shared_ptr. +inline constexpr char kCollectedRowsPointerType[] = "perfetto_collected_rows"; + +// A relation SQLite hands to a pipeline: rows collected into batches. +// +// A column whose type is known holds values of that type or nulls; any other +// column carries its type per row. SQLite's declared types establish nothing, +// so a type is only known when the column traces back to a dataframe column. +class CollectedRows { + public: + CollectedRows(core::Schema columns, StringPool*); + ~CollectedRows(); + CollectedRows(const CollectedRows&) = delete; + CollectedRows& operator=(const CollectedRows&) = delete; + + // Appends a row: one value for each column. Fails on a value a pipeline + // cannot carry, such as a blob, which leaves the rows unusable. + base::Status Append(sqlite3_value** values); + + const core::Schema& columns() const { return columns_; } + uint32_t batch_count() const { + return static_cast(batches_.size()); + } + + // Views batch `i`. The rows must outlive the view. + void View(uint32_t i, core::exec::RowBatch& out) const; + + private: + struct Batch { + std::vector> columns; + // Each column's value buffer, resolved out of its chunk once. + std::vector data; + uint32_t count = 0; + }; + + void AddBatch(); + base::Status Read(Batch&, uint32_t column, sqlite3_value*); + template + base::Status ReadTyped(Batch&, uint32_t column, sqlite3_value*); + + core::Schema columns_; + StringPool* pool_; + std::vector batches_; +}; + +// Reads the rows of the input a run is given at `index`. +class CollectedRowsScan : public core::exec::Source { + public: + // The rows each run reads, one entry per input. Whoever runs the plan sets + // them before each run. + using Inputs = std::vector>; + + // `inputs` must outlive this. + CollectedRowsScan(const Inputs& inputs, uint32_t index); + ~CollectedRowsScan() override; + + std::unique_ptr MakeState() const override; + bool GetData(core::exec::RowBatch& out, + core::exec::OperatorState&) const override; + void Rewind(core::exec::OperatorState&) const override; + base::Status status(const core::exec::OperatorState&) const override; + + private: + struct State : core::exec::OperatorState { + ~State() override; + uint32_t next_batch = 0; + }; + + const Inputs& inputs_; + uint32_t index_; +}; + +} // namespace perfetto::trace_processor::exec + +#endif // SRC_TRACE_PROCESSOR_PERFETTO_SQL_EXEC_COLLECTED_ROWS_H_ diff --git a/src/trace_processor/perfetto_sql/exec/collected_rows_unittest.cc b/src/trace_processor/perfetto_sql/exec/collected_rows_unittest.cc new file mode 100644 index 00000000000..f5e03b0a673 --- /dev/null +++ b/src/trace_processor/perfetto_sql/exec/collected_rows_unittest.cc @@ -0,0 +1,152 @@ +/* + * 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/exec/collected_rows.h" + +#include + +#include +#include +#include +#include +#include + +#include "perfetto/base/status.h" +#include "src/trace_processor/containers/string_pool.h" +#include "src/trace_processor/core/common/schema.h" +#include "src/trace_processor/core/common/storage_types.h" +#include "src/trace_processor/core/exec/row_batch.h" +#include "src/trace_processor/core/exec/row_cursor.h" +#include "src/trace_processor/core/exec/test_utils.h" +#include "src/trace_processor/core/exec/variant.h" +#include "test/gtest_and_gmock.h" + +namespace perfetto::trace_processor::exec { +namespace { + +using core::StorageType; +using core::exec::kMaxBatchRows; +using core::exec::RowBatch; +using core::exec::Variant; +using core::exec::test::ReadNullableColumn; +using testing::ElementsAre; +using testing::HasSubstr; + +class CollectedRowsTest : public testing::Test { + protected: + CollectedRowsTest() { sqlite3_open(":memory:", &db_); } + ~CollectedRowsTest() override { sqlite3_close(db_); } + + // Appends every row of `sql` to `rows`, as SQLite would hand them over. + base::Status Collect(const std::string& sql, CollectedRows& rows) { + sqlite3_stmt* stmt = nullptr; + EXPECT_EQ(sqlite3_prepare_v2(db_, sql.c_str(), -1, &stmt, nullptr), + SQLITE_OK); + base::Status status = base::OkStatus(); + while (status.ok() && sqlite3_step(stmt) == SQLITE_ROW) { + std::vector values; + for (int i = 0; i < sqlite3_column_count(stmt); ++i) { + values.push_back(sqlite3_column_value(stmt, i)); + } + status = rows.Append(values.data()); + } + sqlite3_finalize(stmt); + return status; + } + + StringPool pool_; + sqlite3* db_ = nullptr; +}; + +TEST_F(CollectedRowsTest, TypedColumnsHoldValuesOrNulls) { + CollectedRows rows( + {{"a", StorageType{core::Int64{}}}, {"b", StorageType{core::String{}}}}, + &pool_); + ASSERT_TRUE(Collect("SELECT 1, 'x' UNION ALL SELECT NULL, NULL", rows).ok()); + ASSERT_EQ(rows.batch_count(), 1u); + RowBatch batch; + rows.View(0, batch); + EXPECT_THAT(ReadNullableColumn(batch, 0), + ElementsAre(1, std::nullopt)); + std::vector> b = + ReadNullableColumn(batch, 1); + ASSERT_EQ(b.size(), 2u); + EXPECT_EQ(pool_.Get(*b[0]).ToStdString(), "x"); + EXPECT_FALSE(b[1].has_value()); +} + +TEST_F(CollectedRowsTest, UntypedColumnsCarryTypesPerRow) { + CollectedRows rows({{"v", std::nullopt}}, &pool_); + ASSERT_TRUE(Collect("SELECT 1 UNION ALL SELECT 2.5 UNION ALL SELECT 'x' " + "UNION ALL SELECT NULL", + rows) + .ok()); + RowBatch batch; + rows.View(0, batch); + const auto* values = static_cast(batch.column(0).data()); + EXPECT_EQ(values[0].AsInt64(), 1); + EXPECT_EQ(values[1].AsDouble(), 2.5); + EXPECT_EQ(pool_.Get(values[2].AsString()).ToStdString(), "x"); + EXPECT_EQ(values[3].type, Variant::Type::kNull); +} + +TEST_F(CollectedRowsTest, BlobsCannotBeCarried) { + CollectedRows rows({{"v", std::nullopt}}, &pool_); + EXPECT_THAT(Collect("SELECT X'00'", rows).message(), + HasSubstr("column 'v' holds a blob")); +} + +TEST_F(CollectedRowsTest, TracedTypesMustHold) { + CollectedRows rows({{"a", StorageType{core::Int64{}}}}, &pool_); + EXPECT_THAT(Collect("SELECT 'x'", rows).message(), + HasSubstr("column 'a' does not hold what it was traced back to")); +} + +TEST_F(CollectedRowsTest, RowsFillBatches) { + CollectedRows rows({{"v", StorageType{core::Int64{}}}}, &pool_); + std::string sql = + "WITH RECURSIVE n(v) AS (SELECT 0 UNION ALL SELECT v + 1 " + "FROM n WHERE v + 1 < " + + std::to_string(kMaxBatchRows + 1) + ") SELECT v FROM n"; + ASSERT_TRUE(Collect(sql, rows).ok()); + ASSERT_EQ(rows.batch_count(), 2u); + RowBatch batch; + rows.View(1, batch); + EXPECT_THAT(ReadNullableColumn(batch, 0), + ElementsAre(kMaxBatchRows)); +} + +TEST_F(CollectedRowsTest, ScansReplayTheirInput) { + auto rows = std::make_shared( + core::Schema{{"v", StorageType{core::Int64{}}}}, &pool_); + ASSERT_TRUE(Collect("SELECT 1 UNION ALL SELECT 2", *rows).ok()); + CollectedRowsScan::Inputs inputs{rows}; + CollectedRowsScan scan(inputs, 0); + for (int read = 0; read < 2; ++read) { + core::exec::RowCursor cursor(scan); + ASSERT_TRUE(cursor.Open()); + EXPECT_THAT(ReadNullableColumn(cursor.batch(), 0), + ElementsAre(1, 2)); + uint32_t count = 1; + while (cursor.Next()) { + ++count; + } + EXPECT_EQ(count, 2u); + } +} + +} // namespace +} // namespace perfetto::trace_processor::exec diff --git a/src/trace_processor/perfetto_sql/exec/sql_scan.cc b/src/trace_processor/perfetto_sql/exec/sql_scan.cc deleted file mode 100644 index 7f1bef06570..00000000000 --- a/src/trace_processor/perfetto_sql/exec/sql_scan.cc +++ /dev/null @@ -1,283 +0,0 @@ -/* - * 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/exec/sql_scan.h" - -#include - -#include -#include -#include -#include -#include -#include -#include - -#include "perfetto/base/compiler.h" -#include "perfetto/base/logging.h" -#include "perfetto/base/status.h" -#include "src/trace_processor/containers/string_pool.h" -#include "src/trace_processor/core/common/storage_types.h" -#include "src/trace_processor/core/exec/column_chunk.h" -#include "src/trace_processor/core/exec/column_view.h" -#include "src/trace_processor/core/exec/operator.h" -#include "src/trace_processor/core/exec/row_batch.h" -#include "src/trace_processor/core/exec/row_selection.h" -#include "src/trace_processor/core/exec/variant.h" -#include "src/trace_processor/core/util/bit_vector.h" -#include "src/trace_processor/sqlite/bindings/sqlite_column.h" -#include "src/trace_processor/sqlite/bindings/sqlite_type.h" -#include "src/trace_processor/sqlite/sql_source.h" -#include "src/trace_processor/sqlite/sqlite_connection.h" - -namespace perfetto::trace_processor::exec { -namespace { - -using core::Double; -using core::Int64; -using core::StorageType; -using core::String; -using core::exec::ColumnChunk; -using core::exec::ColumnView; -using core::exec::kMaxBatchRows; -using core::exec::RowBatch; -using core::exec::RowSelection; -using core::exec::Variant; -} // namespace - -SqlScan::SqlScan(SqliteConnection* connection, - SqlSource sql, - core::Schema columns, - StringPool* pool) - : connection_(connection), - sql_(std::move(sql)), - columns_(std::move(columns)), - pool_(pool) {} - -SqlScan::~SqlScan() = default; -SqlScan::State::~State() = default; - -std::unique_ptr SqlScan::MakeState() const { - auto state = std::make_unique(); - Prepare(*state); - return state; -} - -void SqlScan::PrepareColumns(State& state) const { - state.columns.clear(); - state.data.clear(); - state.buffers.resize(columns_.size()); - state.columns.reserve(columns_.size()); - state.data.reserve(columns_.size()); - for (uint32_t i = 0; i < columns_.size(); ++i) { - auto column = state.buffers[i].Acquire(); - void* data = nullptr; - if (!columns_[i].type) { - data = column->Values().data(); - } else { - switch (columns_[i].type->index()) { - case StorageType::GetTypeIndex(): - data = column->Values().data(); - break; - case StorageType::GetTypeIndex(): - data = column->Values().data(); - break; - case StorageType::GetTypeIndex(): - data = column->Values().data(); - break; - case StorageType::GetTypeIndex(): - data = column->Values().data(); - break; - case StorageType::GetTypeIndex(): - data = column->Values().data(); - break; - default: - // An Id was already materialised as a Uint32 by ResolveTypes. - PERFETTO_FATAL("Unreachable"); - } - column->validity.resize(kMaxBatchRows); - } - state.columns.push_back(std::move(column)); - state.data.push_back(data); - } -} - -void SqlScan::Prepare(State& state) const { - state.statement.emplace(connection_->PrepareStatement(sql_)); - state.status = state.statement->status(); - state.done = false; - if (!state.status.ok()) { - return; - } - sqlite3_stmt* stmt = state.statement->sqlite_stmt(); - uint32_t count = sqlite::column::Count(stmt); - if (count != columns_.size()) { - state.status = - base::ErrStatus("SQL source: result shape changed between executions"); - return; - } - for (uint32_t i = 0; i < count; ++i) { - const char* name = sqlite::column::Name(stmt, i); - if (columns_[i].name != (name ? name : "")) { - state.status = base::ErrStatus( - "SQL source: result shape changed between executions"); - return; - } - } -} - -base::Status SqlScan::status(const core::exec::OperatorState& state) const { - return state.Cast().status; -} - -void SqlScan::Rewind(core::exec::OperatorState& state) const { - Prepare(state.Cast()); -} - -bool SqlScan::ReadValue(State& s, - sqlite3_stmt* stmt, - uint32_t index, - uint32_t row) const { - if (!columns_[index].type) { - auto* data = static_cast(s.data[index]); - switch (sqlite::column::Type(stmt, index)) { - case sqlite::Type::kInteger: - data[row] = Variant::Int64(sqlite::column::Int64(stmt, index)); - return true; - case sqlite::Type::kFloat: - data[row] = Variant::Double(sqlite::column::Double(stmt, index)); - return true; - case sqlite::Type::kText: - data[row] = Variant::String( - pool_->InternString(sqlite::column::Text(stmt, index))); - return true; - case sqlite::Type::kNull: - data[row] = Variant::Null(); - return true; - case sqlite::Type::kBlob: - s.status = base::ErrStatus( - "SQL source: column '%s' holds a blob, which a pipeline cannot " - "carry", - columns_[index].name.c_str()); - return false; - } - PERFETTO_FATAL("For GCC"); - } - switch (columns_[index].type->index()) { - case StorageType::GetTypeIndex(): - return ReadTypedValue(s, stmt, index, - row); - case StorageType::GetTypeIndex(): - return ReadTypedValue(s, stmt, index, - row); - case StorageType::GetTypeIndex(): - return ReadTypedValue(s, stmt, index, - row); - case StorageType::GetTypeIndex(): - return ReadTypedValue(s, stmt, index, row); - case StorageType::GetTypeIndex(): - return ReadTypedValue(s, stmt, index, - row); - default: - // An Id was already materialised as a Uint32 by ResolveTypes. - PERFETTO_FATAL("Unreachable"); - } -} - -template -bool SqlScan::ReadTypedValue(State& s, - sqlite3_stmt* stmt, - uint32_t index, - uint32_t row) const { - auto* data = static_cast(s.data[index]); - sqlite::Type type = sqlite::column::Type(stmt, index); - if (PERFETTO_LIKELY(type == SqliteType)) { - if constexpr (std::is_same_v) { - data[row] = pool_->InternString(sqlite::column::Text(stmt, index)); - } else if constexpr (std::is_same_v) { - data[row] = sqlite::column::Double(stmt, index); - } else { - data[row] = static_cast(sqlite::column::Int64(stmt, index)); - } - s.columns[index]->validity.set(row); - return true; - } - if (type == sqlite::Type::kNull) { - // The row is null, but write the slot anyway. A flat column's storage is - // readable at every row, so a reader summing it needs no per-row branch and - // never sees a value left over from the previous batch. - if constexpr (std::is_same_v) { - data[row] = StringPool::Id::Null(); - } else { - data[row] = T{}; - } - return true; - } - // Only reachable if the type lineage established turned out to be wrong. - s.status = base::ErrStatus( - "SQL source: column '%s' does not hold what it was traced back to", - columns_[index].name.c_str()); - return false; -} - -bool SqlScan::GetData(RowBatch& out, core::exec::OperatorState& state) const { - State& s = state.Cast(); - if (s.done || !s.status.ok()) { - return false; - } - out.Reset(); - PrepareColumns(s); - for (const std::shared_ptr& column : s.columns) { - if (column->validity.size() != 0) { - column->validity.ClearAllBits(); - } - } - sqlite3_stmt* stmt = s.statement->sqlite_stmt(); - uint32_t count = 0; - while (count < kMaxBatchRows && s.statement->Step()) { - for (uint32_t i = 0; i < s.columns.size(); ++i) { - if (!ReadValue(s, stmt, i, count)) { - return false; - } - } - ++count; - } - if (!s.statement->status().ok()) { - s.status = s.statement->status(); - return false; - } - s.done = count < kMaxBatchRows; - if (count == 0) { - return false; - } - - for (uint32_t i = 0; i < s.columns.size(); ++i) { - const std::shared_ptr& column = s.columns[i]; - if (!columns_[i].type) { - out.AddColumn( - ColumnView::Variants(static_cast(s.data[i])), column); - } else { - out.AddColumn(ColumnView::Reference(*columns_[i].type, s.data[i], - &column->validity), - column); - } - } - out.Compose(RowSelection::Range(0), count); - out.SetCardinality(count); - return true; -} - -} // namespace perfetto::trace_processor::exec diff --git a/src/trace_processor/perfetto_sql/exec/sql_scan.h b/src/trace_processor/perfetto_sql/exec/sql_scan.h deleted file mode 100644 index 1cd11d95750..00000000000 --- a/src/trace_processor/perfetto_sql/exec/sql_scan.h +++ /dev/null @@ -1,102 +0,0 @@ -/* - * 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_EXEC_SQL_SCAN_H_ -#define SRC_TRACE_PROCESSOR_PERFETTO_SQL_EXEC_SQL_SCAN_H_ - -#include -#include -#include -#include -#include - -#include "perfetto/base/status.h" -#include "src/trace_processor/containers/string_pool.h" -#include "src/trace_processor/core/common/schema.h" -#include "src/trace_processor/core/exec/buffer_pool.h" -#include "src/trace_processor/core/exec/column_chunk.h" -#include "src/trace_processor/core/exec/operator.h" -#include "src/trace_processor/core/exec/row_batch.h" -#include "src/trace_processor/sqlite/bindings/sqlite_type.h" -#include "src/trace_processor/sqlite/sql_source.h" -#include "src/trace_processor/sqlite/sqlite_connection.h" - -struct sqlite3_stmt; - -namespace perfetto::trace_processor::exec { - -// Reads a pipeline's rows from a SQL query. -// -// Promises nothing about the order the rows arrive in, because SQLite does -// not. -// -// Each column carries its type per row unless the query can be traced back to -// a dataframe column, which is the only way to establish a type. SQLite's -// declared types establish nothing: an INTEGER column holds text if something -// puts text in it. -class SqlScan : public core::exec::Source { - public: - // A scan over `sql` with its previously resolved result columns. - SqlScan(SqliteConnection*, SqlSource, core::Schema, StringPool*); - ~SqlScan() override; - - // The query's columns, in the order a batch carries them. - const core::Schema& columns() const { return columns_; } - - // The type of column `i`, or nothing when the column carries a type per - // row. - std::optional column_type(uint32_t i) const { - return columns_[i].type; - } - - std::unique_ptr MakeState() const override; - bool GetData(core::exec::RowBatch& out, - core::exec::OperatorState&) const override; - void Rewind(core::exec::OperatorState&) const override; - base::Status status(const core::exec::OperatorState&) const override; - - private: - struct State : core::exec::OperatorState { - ~State() override; - std::optional statement; - // Shared so a batch can keep the values alive. - std::vector> columns; - std::vector> buffers; - // Each column's value buffer, resolved out of its chunk once. - std::vector data; - bool done = false; - base::Status status = base::OkStatus(); - }; - - void Prepare(State&) const; - void PrepareColumns(State&) const; - - bool ReadValue(State&, sqlite3_stmt*, uint32_t index, uint32_t row) const; - template - bool ReadTypedValue(State&, - sqlite3_stmt*, - uint32_t index, - uint32_t row) const; - - SqliteConnection* connection_; - SqlSource sql_; - core::Schema columns_; - StringPool* pool_; -}; - -} // namespace perfetto::trace_processor::exec - -#endif // SRC_TRACE_PROCESSOR_PERFETTO_SQL_EXEC_SQL_SCAN_H_ diff --git a/src/trace_processor/perfetto_sql/exec/sql_scan_unittest.cc b/src/trace_processor/perfetto_sql/exec/sql_scan_unittest.cc deleted file mode 100644 index 56693cd587a..00000000000 --- a/src/trace_processor/perfetto_sql/exec/sql_scan_unittest.cc +++ /dev/null @@ -1,532 +0,0 @@ -/* - * 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/exec/sql_scan.h" - -#include -#include -#include -#include -#include -#include -#include -#include - -#include "perfetto/ext/base/status_macros.h" -#include "src/trace_processor/containers/string_pool.h" -#include "src/trace_processor/core/common/storage_types.h" -#include "src/trace_processor/core/exec/assert_type.h" -#include "src/trace_processor/core/exec/column_view.h" -#include "src/trace_processor/core/exec/operator.h" -#include "src/trace_processor/core/exec/pipeline.h" -#include "src/trace_processor/core/exec/row_batch.h" -#include "src/trace_processor/core/exec/row_cursor.h" -#include "src/trace_processor/core/exec/row_selection.h" -#include "src/trace_processor/core/exec/test_utils.h" -#include "src/trace_processor/core/exec/tree_accumulate.h" -#include "src/trace_processor/core/exec/tree_number_nodes.h" -#include "src/trace_processor/core/exec/tree_order.h" -#include "src/trace_processor/core/exec/variant.h" -#include "src/trace_processor/core/util/bit_vector.h" -#include "src/trace_processor/perfetto_sql/schema/query_schema.h" -#include "src/trace_processor/perfetto_sql/schema/type_mapping.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::exec { -namespace { - -namespace analysis = ::perfetto::perfetto_sql::analysis; - -using core::BitVector; -using core::Double; -using core::Int64; -using core::StorageType; -using core::String; -using core::exec::ColumnView; -using core::exec::kMaxBatchRows; -using core::exec::RowBatch; -using core::exec::RowCursor; -using core::exec::Variant; - -// Drives a plan the way an executor does: creates the state, owns the batch. -class Execution { - public: - explicit Execution(const core::exec::Source& source) - : source_(source), state_(source.MakeState()) {} - - RowBatch* Next() { - return source_.GetData(batch_, *state_) ? &batch_ : nullptr; - } - void Rewind() { source_.Rewind(*state_); } - base::Status status() const { return source_.status(*state_); } - - private: - const core::exec::Source& source_; - std::unique_ptr state_; - RowBatch batch_; -}; - -std::vector ReadInts(const RowBatch& batch, uint32_t index) { - std::vector out; - for (const Variant& cell : - core::exec::test::ReadColumn(batch, index)) { - out.push_back(cell.AsInt64()); - } - return out; -} - -struct TestColumn { - std::string name; - core::StorageType type; -}; - -TestColumn Typed(std::string name, core::StorageType type) { - return {std::move(name), type}; -} - -class TestCatalog : public analysis::Catalog { - public: - void Add(std::string name, std::vector columns) { - dataframes_[std::move(name)] = std::move(columns); - } - - std::optional FindLeafRelation( - std::string_view name) const override { - auto dataframe = dataframes_.find(std::string(name)); - if (dataframe == dataframes_.end()) { - return std::nullopt; - } - analysis::LeafRelation relation; - relation.name = name; - for (const TestColumn& column : dataframe->second) { - relation.columns.push_back( - {column.name, sql_schema::ToAnalysisType(column.type)}); - } - return relation; - } - - std::optional FindViewSql(std::string_view) const override { - return std::nullopt; - } - - private: - std::map> dataframes_; -}; - -class SqlScanTest : public ::testing::Test { - protected: - SqlScanTest() - : connection_(SqliteConnection::CreateConnectionToNewDatabase()) {} - - void Exec(const std::string& sql) { - auto statement = - connection_->PrepareStatement(SqlSource::FromExecuteQuery(sql)); - ASSERT_TRUE(statement.status().ok()) << statement.status().c_message(); - while (statement.Step()) { - } - ASSERT_TRUE(statement.status().ok()) << statement.status().c_message(); - } - - base::StatusOr> Scan( - const std::string& sql, - const analysis::Catalog& catalog) { - auto source = SqlSource::FromExecuteQuery(sql); - ASSIGN_OR_RETURN(auto columns, sql_schema::DescribeQuery(connection_.get(), - source, catalog)); - return std::make_unique(connection_.get(), std::move(source), - std::move(columns), &pool_); - } - - // An empty catalog traces nothing, so every column is a variant. - base::StatusOr> Scan(const std::string& sql) { - return Scan(sql, empty_catalog_); - } - - StringPool pool_; - std::unique_ptr connection_; - TestCatalog empty_catalog_; -}; - -TEST_F(SqlScanTest, AQuerysColumnsAreKnownBeforeItsRows) { - auto scan = Scan("SELECT 1 AS a, 'x' AS b"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - ASSERT_EQ((*scan)->columns().size(), 2u); - EXPECT_EQ((*scan)->columns()[0].name, "a"); - EXPECT_EQ((*scan)->columns()[1].name, "b"); -} - -TEST_F(SqlScanTest, AQuerysRowsArriveAsABatch) { - auto scan = Scan( - "SELECT a FROM (" - "SELECT 2 AS ord, 20 AS a UNION ALL " - "SELECT 3, 30 UNION ALL SELECT 1, 10) ORDER BY ord"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Execution run(**scan); - - RowBatch* batch = run.Next(); - ASSERT_NE(batch, nullptr); - EXPECT_EQ(batch->size(), 3u); - EXPECT_THAT(ReadInts(*batch, 0), testing::ElementsAre(10, 20, 30)); - EXPECT_EQ(run.Next(), nullptr); - EXPECT_TRUE(run.status().ok()); -} - -// One column holding three different types and a null, which SQLite allows. -TEST_F(SqlScanTest, OneColumnCanHoldMoreThanOneType) { - auto scan = Scan( - "SELECT value FROM (" - "SELECT 1 AS ord, 7 AS value UNION ALL SELECT 2, 1.5 " - "UNION ALL SELECT 3, 'hello' UNION ALL SELECT 4, NULL) ORDER BY ord"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Execution run(**scan); - - RowBatch* batch = run.Next(); - ASSERT_NE(batch, nullptr); - std::vector cells = core::exec::test::ReadColumn(*batch, 0); - ASSERT_EQ(cells.size(), 4u); - EXPECT_EQ(cells[0].AsInt64(), 7); - EXPECT_EQ(cells[1].AsDouble(), 1.5); - EXPECT_EQ(pool_.Get(cells[2].AsString()).ToStdString(), "hello"); - EXPECT_EQ(cells[3].type, Variant::Type::kNull); - EXPECT_TRUE(run.status().ok()); -} - -// A declared type is not binding in SQLite, so it is not trusted here. -TEST_F(SqlScanTest, ADeclaredTypeIsNotBelieved) { - Exec("CREATE TABLE t(i INTEGER)"); - Exec("INSERT INTO t VALUES(1), ('not a number')"); - auto scan = Scan("SELECT i FROM t"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Execution run(**scan); - - RowBatch* batch = run.Next(); - ASSERT_NE(batch, nullptr); - std::vector cells = core::exec::test::ReadColumn(*batch, 0); - ASSERT_EQ(cells.size(), 2u); - EXPECT_EQ(cells[0].AsInt64(), 1); - EXPECT_EQ(pool_.Get(cells[1].AsString()).ToStdString(), "not a number"); - EXPECT_TRUE(run.status().ok()); -} - -TEST_F(SqlScanTest, AColumnWhichIsNeverAnythingIsAColumnOfNulls) { - auto scan = Scan("SELECT NULL AS a UNION ALL SELECT NULL"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Execution run(**scan); - - RowBatch* batch = run.Next(); - ASSERT_NE(batch, nullptr); - for (const Variant& cell : core::exec::test::ReadColumn(*batch, 0)) { - EXPECT_EQ(cell.type, Variant::Type::kNull); - } -} - -TEST_F(SqlScanTest, MoreRowsThanFitInABatchArriveInSeveral) { - Exec( - "CREATE TABLE t AS WITH RECURSIVE r(x) AS (" - " SELECT 0 UNION ALL SELECT x + 1 FROM r WHERE x < 4999" - ") SELECT x FROM r"); - auto scan = Scan("SELECT x FROM t ORDER BY x"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Execution run(**scan); - - std::vector seen; - std::vector sizes; - while (RowBatch* batch = run.Next()) { - sizes.push_back(batch->size()); - for (int64_t value : ReadInts(*batch, 0)) { - seen.push_back(value); - } - } - ASSERT_TRUE(run.status().ok()) << run.status().c_message(); - EXPECT_THAT(sizes, testing::ElementsAre(kMaxBatchRows, kMaxBatchRows, 904u)); - ASSERT_EQ(seen.size(), 5000u); - for (uint32_t i = 0; i < seen.size(); ++i) { - ASSERT_EQ(seen[i], int64_t{i}); - } -} - -TEST_F(SqlScanTest, AQueryWhichDoesNotRunIsReported) { - auto scan = Scan("SELECT * FROM not_a_table"); - EXPECT_FALSE(scan.ok()); -} - -TEST_F(SqlScanTest, AQueryWhichFailsPartWayThroughIsReported) { - auto scan = - Scan("SELECT 1 AS a UNION ALL SELECT abs(-9223372036854775807 - 1)"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Execution run(**scan); - while (run.Next()) { - } - EXPECT_FALSE(run.status().ok()); -} - -TEST_F(SqlScanTest, ABlobIsReportedRatherThanCarried) { - auto scan = Scan("SELECT x'0102' AS a"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Execution run(**scan); - - EXPECT_EQ(run.Next(), nullptr); - EXPECT_FALSE(run.status().ok()); - EXPECT_THAT(run.status().message(), testing::HasSubstr("blob")); -} - -TEST_F(SqlScanTest, RewindClearsThePreviousExecutionError) { - auto scan = Scan("SELECT x'0102' AS a"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Execution run(**scan); - ASSERT_EQ(run.Next(), nullptr); - ASSERT_FALSE(run.status().ok()); - - run.Rewind(); - EXPECT_TRUE(run.status().ok()); - EXPECT_EQ(run.Next(), nullptr); - EXPECT_FALSE(run.status().ok()); -} - -TEST_F(SqlScanTest, EachExecutionValidatesItsResultShape) { - Exec("CREATE TABLE t(a INTEGER)"); - Exec("INSERT INTO t VALUES(1)"); - auto scan = Scan("SELECT * FROM t"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Exec("ALTER TABLE t ADD COLUMN b TEXT"); - - Execution run(**scan); - EXPECT_EQ(run.Next(), nullptr); - EXPECT_FALSE(run.status().ok()); - EXPECT_THAT(run.status().message(), testing::HasSubstr("shape changed")); -} - -TEST_F(SqlScanTest, AScanCanBeRunAgain) { - auto scan = Scan("SELECT 1 AS a UNION ALL SELECT 2"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Execution run(**scan); - - auto drain = [&] { - std::vector out; - while (RowBatch* batch = run.Next()) { - for (int64_t value : ReadInts(*batch, 0)) { - out.push_back(value); - } - } - return out; - }; - std::vector first = drain(); - run.Rewind(); - EXPECT_EQ(drain(), first); -} - -TEST_F(SqlScanTest, AQueryReachesARowCursor) { - auto scan = Scan( - "SELECT a FROM (SELECT 2 AS ord, 6 AS a UNION ALL " - "SELECT 3, 7 UNION ALL SELECT 1, 5) ORDER BY ord"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - - RowCursor cursor(**scan); - std::vector values; - for (bool more = cursor.Open(); more; more = cursor.Next()) { - values.push_back(cursor.Value(0).AsInt64()); - } - EXPECT_THAT(values, testing::ElementsAre(5, 6, 7)); -} - -// The whole pipeline: a query, its columns asserted to be integers, put into a -// tree order and folded up the tree. -TEST_F(SqlScanTest, AQueryReachesTheTreeOperators) { - Exec( - "CREATE TABLE t AS " - "SELECT 0 AS id, NULL AS parent_id, 10 AS self " - "UNION ALL SELECT 1, 0, 20 " - "UNION ALL SELECT 2, 0, 30 " - "UNION ALL SELECT 3, 1, 40"); - auto scan = Scan("SELECT id, parent_id, self FROM t ORDER BY id"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - - std::vector> ops; - ops.push_back(std::make_unique( - 0, core::exec::AssertTypeTarget{core::Int64{}}, "id")); - ops.push_back(std::make_unique( - 1, core::exec::AssertTypeTarget{core::Int64{}}, "parent_id")); - ops.push_back(std::make_unique( - 2, core::exec::AssertTypeTarget{core::Int64{}}, "self")); - ops.push_back(std::make_unique(0, 1)); - ops.push_back(std::make_unique(3, 4)); - core::exec::TreeAccumulateSpec spec{3, 4, 2}; - ops.push_back(std::make_unique(spec)); - core::exec::Pipeline folded(**scan, std::move(ops)); - - std::unique_ptr state = folded.MakeState(); - RowBatch batch; - std::vector totals(4, 0); - while (folded.GetData(batch, *state)) { - std::vector ids = core::exec::test::ReadColumn(batch, 0); - std::vector values = - core::exec::test::ReadColumn(batch, 5); - for (uint32_t row = 0; row < batch.size(); ++row) { - totals[static_cast(ids[row])] = values[row]; - } - } - ASSERT_TRUE(folded.status(*state).ok()) << folded.status(*state).message(); - EXPECT_THAT(totals, testing::ElementsAre(100, 60, 30, 40)); -} - -// A column which is not the type claimed for it fails the pipeline. -TEST_F(SqlScanTest, AColumnWhichIsNotWhatWasAssertedIsReported) { - Exec("CREATE TABLE t(i INTEGER)"); - Exec("INSERT INTO t VALUES(1), ('not a number')"); - auto scan = Scan("SELECT i FROM t"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - - std::vector> ops; - ops.push_back(std::make_unique( - 0, core::exec::AssertTypeTarget{core::Int64{}}, "i")); - core::exec::Pipeline typed(**scan, std::move(ops)); - - Execution run(typed); - while (run.Next()) { - } - EXPECT_FALSE(run.status().ok()); - EXPECT_THAT(run.status().message(), testing::HasSubstr("'i'")); -} - -// A column which can be traced back to a dataframe needs neither a variant nor -// an assertion: it comes out flat. -TEST_F(SqlScanTest, AColumnFollowedBackToADataframeComesOutFlat) { - Exec("CREATE TABLE df(id INTEGER, name TEXT)"); - Exec("INSERT INTO df VALUES(7, 'hello'), (8, NULL)"); - TestCatalog catalog; - catalog.Add("df", {Typed("id", core::StorageType{core::Int64{}}), - Typed("name", core::StorageType{core::String{}})}); - - auto scan = Scan("SELECT id, name FROM df ORDER BY id", catalog); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - ASSERT_TRUE((*scan)->column_type(0).has_value()); - EXPECT_TRUE((*scan)->column_type(0)->Is()); - EXPECT_TRUE((*scan)->column_type(1)->Is()); - - Execution run(**scan); - RowBatch* batch = run.Next(); - ASSERT_NE(batch, nullptr); - EXPECT_EQ(batch->column(0).kind(), ColumnView::Kind::kFlat); - const auto* ids = static_cast(batch->column(0).data()); - EXPECT_EQ(ids[0], 7); - EXPECT_EQ(ids[1], 8); - const BitVector* validity = batch->column(1).validity(); - ASSERT_NE(validity, nullptr); - EXPECT_TRUE(validity->is_set(0)); - EXPECT_FALSE(validity->is_set(1)); - std::vector names = - core::exec::test::ReadColumn(*batch, 1); - EXPECT_EQ(pool_.Get(names[0]).ToStdString(), "hello"); -} - -TEST_F(SqlScanTest, MixedNumericCompoundResultsStayVariants) { - Exec("CREATE TABLE ints(value INTEGER)"); - Exec("INSERT INTO ints VALUES(7)"); - Exec("CREATE TABLE doubles(value REAL)"); - Exec("INSERT INTO doubles VALUES(1.5)"); - TestCatalog catalog; - catalog.Add("ints", {Typed("value", StorageType{Int64{}})}); - catalog.Add("doubles", {Typed("value", StorageType{Double{}})}); - - auto scan = Scan( - "SELECT value FROM (" - "SELECT 1 AS ord, value FROM ints UNION ALL " - "SELECT 2, value FROM doubles) ORDER BY ord", - catalog); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - EXPECT_FALSE((*scan)->column_type(0).has_value()); - Execution run(**scan); - RowBatch* batch = run.Next(); - ASSERT_NE(batch, nullptr); - std::vector values = - core::exec::test::ReadColumn(*batch, 0); - ASSERT_EQ(values.size(), 2u); - EXPECT_EQ(values[0].AsInt64(), 7); - EXPECT_EQ(values[1].AsDouble(), 1.5); -} - -// An expression cannot be traced back, so it stays a variant even when the -// column beside it does not. -TEST_F(SqlScanTest, OnlyTheColumnsWhichCanBeFollowedComeOutFlat) { - Exec("CREATE TABLE df(id INTEGER)"); - Exec("INSERT INTO df VALUES(7)"); - TestCatalog catalog; - catalog.Add("df", {Typed("id", core::StorageType{core::Int64{}})}); - - auto scan = Scan("SELECT id, id * 2 AS doubled FROM df", catalog); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - EXPECT_TRUE((*scan)->column_type(0).has_value()); - EXPECT_FALSE((*scan)->column_type(1).has_value()); - - Execution run(**scan); - RowBatch* batch = run.Next(); - ASSERT_NE(batch, nullptr); - EXPECT_EQ(batch->column(0).kind(), ColumnView::Kind::kFlat); - EXPECT_EQ(batch->column(1).kind(), ColumnView::Kind::kVariant); -} - -// Without a catalog nothing can be traced back. -TEST_F(SqlScanTest, WithoutACatalogEveryColumnIsAVariant) { - Exec("CREATE TABLE df(id INTEGER)"); - auto scan = Scan("SELECT id FROM df"); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - EXPECT_FALSE((*scan)->column_type(0).has_value()); -} - -// A flat column's storage is readable at every row, so a reader which sums it -// without checking validity gets zero rather than a value left over from the -// previous batch. -TEST_F(SqlScanTest, ANullSlotOfAFlatColumnHoldsZero) { - Exec("CREATE TABLE df(id INTEGER)"); - Exec("INSERT INTO df VALUES(7), (NULL)"); - TestCatalog catalog; - catalog.Add("df", {Typed("id", core::StorageType{core::Int64{}})}); - - auto scan = Scan("SELECT id FROM df ORDER BY id IS NULL", catalog); - ASSERT_TRUE(scan.ok()) << scan.status().c_message(); - Execution run(**scan); - RowBatch* batch = run.Next(); - ASSERT_NE(batch, nullptr); - ASSERT_EQ(batch->size(), 2u); - - const auto* ids = static_cast(batch->column(0).data()); - EXPECT_EQ(ids[0], 7); - EXPECT_EQ(ids[1], 0); - EXPECT_FALSE(batch->column(0).validity()->is_set(1)); -} - -TEST_F(SqlScanTest, RetainedBatchSurvivesAdvanceAndRewind) { - auto scan = Scan( - "WITH RECURSIVE n(x) AS (SELECT 0 UNION ALL " - "SELECT x+1 FROM n WHERE x<4096) SELECT x FROM n"); - ASSERT_TRUE(scan.ok()) << scan.status().message(); - auto state = (*scan)->MakeState(); - RowBatch output, retained; - ASSERT_TRUE((*scan)->GetData(output, *state)); - retained.CopyFrom(output); - ASSERT_TRUE((*scan)->GetData(output, *state)); - (*scan)->Rewind(*state); - ASSERT_TRUE((*scan)->GetData(output, *state)); - auto values = ReadInts(retained, 0); - ASSERT_EQ(values.size(), kMaxBatchRows); - for (uint32_t i = 0; i < values.size(); ++i) - EXPECT_EQ(values[i], i); -} - -} // namespace -} // namespace perfetto::trace_processor::exec diff --git a/src/trace_processor/perfetto_sql/parser/BUILD.gn b/src/trace_processor/perfetto_sql/parser/BUILD.gn index 2a855572a37..7932e863dbe 100644 --- a/src/trace_processor/perfetto_sql/parser/BUILD.gn +++ b/src/trace_processor/perfetto_sql/parser/BUILD.gn @@ -29,10 +29,12 @@ source_set("parser") { "../../../../gn:sqlite", "../../../base", "../../../perfetto_sql:intrinsic_macro_expansion", + "../../../perfetto_sql/analysis", "../../../perfetto_sql/syntaqlite", "../../sqlite", "../../util:sql_argument", "../pipeline:logical", + "../schema", ] cflags = [ "-Wno-old-style-cast", diff --git a/src/trace_processor/perfetto_sql/parser/perfetto_sql_parser.cc b/src/trace_processor/perfetto_sql/parser/perfetto_sql_parser.cc index 02ed414507b..89a513956ad 100644 --- a/src/trace_processor/perfetto_sql/parser/perfetto_sql_parser.cc +++ b/src/trace_processor/perfetto_sql/parser/perfetto_sql_parser.cc @@ -33,6 +33,7 @@ #include "perfetto/ext/base/status_or.h" #include "perfetto/ext/base/string_utils.h" #include "perfetto/ext/base/string_view.h" +#include "src/perfetto_sql/analysis/relation.h" #include "src/perfetto_sql/intrinsic_macro_expansion.h" #include "src/perfetto_sql/syntaqlite/syntaqlite_perfetto.h" #include "src/trace_processor/perfetto_sql/parser/function_util.h" @@ -40,6 +41,7 @@ #include "src/trace_processor/perfetto_sql/pipeline/compiler.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" #include "src/trace_processor/perfetto_sql/pipeline/pipeline_sql.h" +#include "src/trace_processor/perfetto_sql/schema/type_mapping.h" #include "src/trace_processor/sqlite/sql_source.h" #include "src/trace_processor/util/sql_argument.h" @@ -114,6 +116,8 @@ base::StatusOr BuildReturnType( return result; } +namespace analysis = ::perfetto::perfetto_sql::analysis; + // --------------------------------------------------------------------------- // Rewrite tree -> SqlSource // --------------------------------------------------------------------------- @@ -402,6 +406,53 @@ class RewriteTree { std::vector source_rooted_; }; +// The catalog, plus what the current statement's pipelines output: SQL which +// reads a pipeline, such as another pipeline's source, is described by the +// plan the pipeline compiled to. +class StatementCatalog final : public pipeline::Catalog { + public: + explicit StatementCatalog(const pipeline::Catalog& catalog) + : catalog_(catalog) {} + + // Forgets the pipelines of the previous statement. + void Reset() { pipelines_.Clear(); } + + void AddPipeline(uint32_t node, const pipeline::LogicalPlan& plan) { + analysis::LeafRelation relation; + relation.name = "pipeline"; + for (const pipeline::NamedColumn& column : plan.output) { + std::optional type = plan.columns[column.id].type; + relation.columns.push_back( + {column.name, + type ? std::make_optional(sql_schema::ToAnalysisType(*type)) + : std::nullopt, + false}); + } + pipelines_.Insert(node, std::move(relation)); + } + + std::optional FindLeafRelation( + std::string_view name) const override { + return catalog_.FindLeafRelation(name); + } + std::optional FindViewSql(std::string_view name) const override { + return catalog_.FindViewSql(name); + } + std::optional FindNodeRelation( + analysis::SqlNode node) const override { + const analysis::LeafRelation* relation = pipelines_.Find(node.id); + return relation ? std::make_optional(*relation) : std::nullopt; + } + const dataframe::Dataframe* FindDataframe( + std::string_view name) const override { + return catalog_.FindDataframe(name); + } + + private: + const pipeline::Catalog& catalog_; + base::FlatHashMap pipelines_; +}; + SqlSource NodeSource(const RewriteTree& rb, uint32_t node_id) { auto s = rb.NodeSource(node_id); PERFETTO_CHECK(s.has_value()); @@ -442,10 +493,11 @@ base::StatusOr CompilePipeline( // SQL is expected is replaced with. base::StatusOr PipelineSql(SyntaqliteParser* p, const RewriteTree& rb, - const pipeline::Catalog* catalog, + StatementCatalog* catalog, uint32_t pipeline_id) { ASSIGN_OR_RETURN(pipeline::LogicalPlan plan, CompilePipeline(p, rb, catalog, pipeline_id)); + catalog->AddPipeline(pipeline_id, plan); base::StatusOr sql = pipeline::SelectPipelineSql(plan); if (!sql.ok()) { return base::ErrStatus("%s%s", @@ -692,7 +744,7 @@ struct PerfettoSqlParser::Impl { bool allowed) : source(SqlSource::FromTraceProcessorImplementation("")), macros(m), - catalog(&c), + statement_catalog(c), pipelines_allowed(allowed) { synq = syntaqlite_parser_create_perfetto(nullptr); PERFETTO_CHECK(synq != nullptr); @@ -778,8 +830,8 @@ struct PerfettoSqlParser::Impl { // Replaces the pipeline at `node` with SQL reading it. int ExpandPipeline(SyntaqliteParser* parser, uint32_t node) { RewriteTree rb(parser, source, CurrentStatementDocOffset(parser), macros); - base::StatusOr sql = - PipelineSql(parser, rb, pipelines_allowed ? catalog : nullptr, node); + base::StatusOr sql = PipelineSql( + parser, rb, pipelines_allowed ? &statement_catalog : nullptr, node); if (!sql.ok()) { expansion_error = sql.status(); return SYNTAQLITE_NODE_EXPAND_ERROR; @@ -792,7 +844,7 @@ struct PerfettoSqlParser::Impl { SyntaqliteParser* synq; SqlSource source; const base::FlatHashMap& macros; - const pipeline::Catalog* catalog; + StatementCatalog statement_catalog; // Whether the SQL being read may use a pipeline. Set per source, so one // parser serves both the standard library and user SQL. bool pipelines_allowed; @@ -813,6 +865,7 @@ bool PerfettoSqlParser::Impl::Next( current_statement = std::nullopt; out_statement_sql = std::nullopt; + statement_catalog.Reset(); const SqlSource& stmt = source; uint32_t root = 0; @@ -853,9 +906,9 @@ bool PerfettoSqlParser::Impl::Next( const auto* node = static_cast(syntaqlite_parser_node(synq, root)); - auto result = - ParseStatement(synq, rb, stmt, stmt_doc_offset, - pipelines_allowed ? catalog : nullptr, root, node); + auto result = ParseStatement(synq, rb, stmt, stmt_doc_offset, + pipelines_allowed ? &statement_catalog : nullptr, + root, node); if (!result.ok()) { status = result.status(); return false; diff --git a/src/trace_processor/perfetto_sql/parser/perfetto_sql_parser_unittest.cc b/src/trace_processor/perfetto_sql/parser/perfetto_sql_parser_unittest.cc index 2c4c874f5bc..c23391c82e4 100644 --- a/src/trace_processor/perfetto_sql/parser/perfetto_sql_parser_unittest.cc +++ b/src/trace_processor/perfetto_sql/parser/perfetto_sql_parser_unittest.cc @@ -31,6 +31,7 @@ #include "src/trace_processor/perfetto_sql/parser/function_util.h" #include "src/trace_processor/perfetto_sql/parser/perfetto_sql_test_utils.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" +#include "src/trace_processor/perfetto_sql/pipeline/pipeline_sql.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" @@ -615,7 +616,7 @@ TEST_F(PerfettoSqlParserTest, PipelineExpandsMacros) { "FROM (SELECT * FROM tree) |> TREE ACCUMULATE UP SUM(self) AS " "total"); EXPECT_THAT(pipeline::LogicalPlanToString(pipeline->plan), - HasSubstr("Scan(sql SELECT * FROM (SELECT * FROM tree))")); + HasSubstr("Scan(sql (SELECT * FROM tree))")); } TEST_F(PerfettoSqlParserTest, CreatePerfettoTableAsPipeline) { @@ -655,7 +656,7 @@ TEST_F(PerfettoSqlParserTest, PipelineSyntaxErrors) { TEST_F(PerfettoSqlParserTest, PipelineCompileErrors) { EXPECT_THAT(ParsePipeline("FROM nope").status().message(), - HasSubstr("no such table: nope")); + HasSubstr("'nope' is not known")); EXPECT_THAT(ParsePipeline("FROM (SELECT id, self FROM tree) |> TREE " "ACCUMULATE UP SUM(self) AS total") .status() @@ -863,27 +864,21 @@ TEST_F(PerfettoSqlParserTest, PipelineSkipsUnusedTreeAggregates) { } TEST_F(PerfettoSqlParserTest, PipelinePushesPruningIntoSql) { - // SQL sources only ask SQLite for the columns that are used. - auto plan = ParsePipeline( + // SQL sources only collect the columns that are used. + PerfettoSqlParser parser(macros_, catalog_, /*pipelines_allowed=*/true); + parser.Reset(SqlSource::FromExecuteQuery( "INTERVAL INTERSECTION OF (sql_spans AS a, (SELECT * FROM sql_spans) AS " - "b) |> SELECT ts, dur, b.utid"); - ASSERT_TRUE(plan.ok()) << plan.status().message(); - EXPECT_THAT(*plan, - HasSubstr("Scan(sql WITH __pipeline_source(c0, c1, c2, c3) AS " - "(SELECT * FROM sql_spans AS a) SELECT c0 AS \"ts\", " - "c1 AS \"dur\" FROM __pipeline_source) " - "[#2 AS ts, #3 AS dur]")); - EXPECT_THAT( - *plan, - HasSubstr("Scan(sql WITH __pipeline_source(c0, c1, c2, c3) AS " - "(SELECT * FROM (SELECT * FROM sql_spans) AS b) SELECT c0 AS " - "\"ts\", c1 AS \"dur\", c3 AS \"utid\" FROM __pipeline_source) " - "[#6 AS ts, #7 AS dur, #9 AS utid]")); - - // A source that uses every column is left alone. - plan = ParsePipeline("FROM tree |> TREE ACCUMULATE UP SUM(self) AS total"); - ASSERT_TRUE(plan.ok()) << plan.status().message(); - EXPECT_THAT(*plan, HasSubstr("Scan(sql SELECT * FROM tree)")); + "b) |> SELECT ts, dur, b.utid")); + ASSERT_TRUE(parser.Next()) << parser.status().message(); + auto sql = + pipeline::SelectPipelineSql(std::get(parser.statement()).plan); + ASSERT_TRUE(sql.ok()) << sql.status().message(); + EXPECT_THAT(*sql, HasSubstr("(SELECT __intrinsic_rows('v:ts,v:dur', \"ts\", " + "\"dur\") FROM sql_spans AS a)")); + EXPECT_THAT(*sql, + HasSubstr("(SELECT __intrinsic_rows('v:ts,v:dur,v:utid', \"ts\", " + "\"dur\", \"utid\") FROM (SELECT * FROM sql_spans) AS " + "b)")); } // The relational operators which reshape a pipeline's row: EXTEND, DROP, @@ -1097,29 +1092,27 @@ TEST_F(PerfettoSqlParserSelectLikeTest, SourceNames) { Check({ {"FROM (SELECT 1 AS x, count(*) AS n FROM tree) AS t |> SELECT *", "Output(#0 AS x, #1 AS n)"}, - // SQLite names an unaliased expression after its text. + // An expression is only named by its alias. {"FROM (SELECT 1 AS x, 1 + 1 FROM tree) AS t |> SELECT x", - "expected every column to have a valid name, but '1 + 1' is not one: " - "give it one with AS"}, + "expected every column to have a name, but column 2 has none: give it " + "one with AS"}, {"FROM (SELECT 1 AS \"my col\") AS t", kInvalid}, {"FROM (SELECT 1 AS \"1x\") AS t", kInvalid}, // Without an earlier x, x:1 is simply not a valid name. {"FROM (SELECT 1 AS \"x:1\") AS t", kInvalid}, {"INTERVAL INTERSECTION OF ((SELECT 0 AS ts, 10 AS dur, 1 + 1) AS a, " "(SELECT 5 AS ts, 10 AS dur) AS b)", - kInvalid}, + "expected every column to have a name, but column 3 has none"}, }); } -// SQLite renames the second of two columns sharing a name `x` to `x:1`. That -// is reported as the duplicate it is, before the names are checked. +// Names are compared as SQL compares them, ignoring case. TEST_F(PerfettoSqlParserSelectLikeTest, SourceDuplicateNames) { Check({ {"FROM (SELECT 1 AS x, 2 AS x) AS t", - "expected distinct column names, but there are two named 'x', which " - "SQLite renamed to 'x' and 'x:1'"}, + "expected distinct column names, but there are two named 'x'"}, {"FROM (SELECT 1 AS x, 2 AS X) AS t", - "two named 'x', which SQLite renamed to 'x' and 'X:1'"}, + "expected distinct column names, but there are two named 'x'"}, {"INTERVAL INTERSECTION OF (" "(SELECT 0 AS ts, 10 AS dur, 1 AS x, 2 AS x) AS a, " "(SELECT 5 AS ts, 10 AS dur) AS b)", diff --git a/src/trace_processor/perfetto_sql/pipeline/BUILD.gn b/src/trace_processor/perfetto_sql/pipeline/BUILD.gn index 7b169d55c32..bc290babc71 100644 --- a/src/trace_processor/perfetto_sql/pipeline/BUILD.gn +++ b/src/trace_processor/perfetto_sql/pipeline/BUILD.gn @@ -34,11 +34,13 @@ source_set("logical") { deps = [ "../../../../gn:default_deps", "../../../base", + "../../../perfetto_sql/analysis", "../../../perfetto_sql/syntaqlite", "../../core/common", "../../core/dataframe", "../../sqlite", "../../util:sql_argument", + "../schema", ] } diff --git a/src/trace_processor/perfetto_sql/pipeline/catalog.h b/src/trace_processor/perfetto_sql/pipeline/catalog.h index 8c381417518..a128a68b780 100644 --- a/src/trace_processor/perfetto_sql/pipeline/catalog.h +++ b/src/trace_processor/perfetto_sql/pipeline/catalog.h @@ -19,26 +19,21 @@ #include -#include "perfetto/ext/base/status_or.h" +#include "src/perfetto_sql/analysis/relation.h" #include "src/trace_processor/core/dataframe/dataframe.h" -#include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" -#include "src/trace_processor/sqlite/sql_source.h" namespace perfetto::trace_processor::pipeline { -// Lookup interface the compiler uses to resolve what a pipeline reads. -class Catalog { +// Lookup interface the compiler uses to resolve what a pipeline reads: the +// relations semantic analysis can describe, and the dataframes a pipeline can +// read directly. +class Catalog : public perfetto_sql::analysis::Catalog { public: - virtual ~Catalog(); + ~Catalog() override; // Dataframe registered as `name`, or null. virtual const dataframe::Dataframe* FindDataframe( std::string_view name) const = 0; - - // Columns of the query `sql`, typed where they trace back to a dataframe - // column. Fails if SQLite cannot prepare the query, or if it reads a - // parameter, which nothing binds when a pipeline runs it. - virtual base::StatusOr DescribeQuery(const SqlSource& sql) const = 0; }; } // namespace perfetto::trace_processor::pipeline diff --git a/src/trace_processor/perfetto_sql/pipeline/column_pruning.cc b/src/trace_processor/perfetto_sql/pipeline/column_pruning.cc index 6e62e523252..6a6e41dd966 100644 --- a/src/trace_processor/perfetto_sql/pipeline/column_pruning.cc +++ b/src/trace_processor/perfetto_sql/pipeline/column_pruning.cc @@ -25,7 +25,6 @@ #include #include "perfetto/base/logging.h" -#include "perfetto/ext/base/string_utils.h" #include "perfetto/ext/base/variant.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" #include "src/trace_processor/sqlite/sql_source.h" @@ -36,35 +35,6 @@ namespace { // The columns something downstream uses, indexed by column ID. using Needed = std::vector; -// Narrows `sql` down to the columns at the positions in `keep`, which keep -// their names from `kept`. We pick by position rather than name because a -// query can return two columns with the same name. The scan checks the names -// it reads back, so each column is given its name again. -SqlSource SelectPositions(const SqlSource& sql, - uint32_t count, - const std::vector& keep, - const std::vector& kept) { - std::string names; - for (uint32_t i = 0; i < count; ++i) { - if (i) { - names += ", "; - } - names += "c" + std::to_string(i); - } - std::string selected; - for (uint32_t i = 0; i < keep.size(); ++i) { - if (i) { - selected += ", "; - } - selected += "c" + std::to_string(keep[i]) + " AS \"" + - base::ReplaceAll(kept[i].name, "\"", "\"\"") + "\""; - } - return sql.RewriteAllIgnoreExisting( - SqlSource::FromTraceProcessorImplementation( - "WITH __pipeline_source(" + names + ") AS (" + sql.sql() + - ") SELECT " + selected + " FROM __pipeline_source")); -} - void PruneScan(op::Scan& scan, const Needed& needed) { std::vector keep; for (uint32_t i = 0; i < scan.columns.size(); ++i) { @@ -79,7 +49,6 @@ void PruneScan(op::Scan& scan, const Needed& needed) { if (keep.empty()) { keep.push_back(0); } - auto count = static_cast(scan.columns.size()); std::vector columns; for (uint32_t i : keep) { columns.push_back(std::move(scan.columns[i])); @@ -97,11 +66,9 @@ void PruneScan(op::Scan& scan, const Needed& needed) { dataframe.columns = std::move(kept); return; } - case Kind::GetTypeIndex(): { - auto& sql = base::unchecked_get(scan.source); - sql = SelectPositions(sql, count, keep, scan.columns); + case Kind::GetTypeIndex(): + // The SQL collecting the relation reads only the columns the scan keeps. return; - } default: PERFETTO_FATAL("Unknown scan source"); } diff --git a/src/trace_processor/perfetto_sql/pipeline/compiler.cc b/src/trace_processor/perfetto_sql/pipeline/compiler.cc index afff6750907..3cd6f1fde45 100644 --- a/src/trace_processor/perfetto_sql/pipeline/compiler.cc +++ b/src/trace_processor/perfetto_sql/pipeline/compiler.cc @@ -31,18 +31,23 @@ #include "perfetto/ext/base/status_or.h" #include "perfetto/ext/base/string_utils.h" #include "perfetto/ext/base/string_view.h" +#include "src/perfetto_sql/analysis/relation.h" #include "src/perfetto_sql/syntaqlite/syntaqlite_perfetto.h" #include "src/trace_processor/core/common/storage_types.h" #include "src/trace_processor/core/dataframe/dataframe.h" #include "src/trace_processor/perfetto_sql/pipeline/catalog.h" #include "src/trace_processor/perfetto_sql/pipeline/column_pruning.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" +#include "src/trace_processor/perfetto_sql/schema/type_mapping.h" #include "src/trace_processor/sqlite/sql_source.h" #include "src/trace_processor/util/sql_argument.h" namespace perfetto::trace_processor::pipeline { namespace { +namespace analysis = ::perfetto::perfetto_sql::analysis; +using core::StorageType; + // The name `span` spells. A quoted name escapes its closing quote by doubling // it, which the span, pointing into the source, still contains. std::string SpanText(SyntaqliteParser* p, SyntaqliteTextSpan span) { @@ -374,28 +379,16 @@ base::Status Compiler::CheckSourceNames(const std::vector& columns, } } } - // SQLite renames the second of two columns sharing a name `x` to `x:1`. - // Say so, rather than only that `x:1` is not a valid name. for (size_t i = 0; i < columns.size(); ++i) { const std::string& name = columns[i].name; - size_t colon = name.rfind(':'); - if (colon == std::string::npos || colon + 1 == name.size() || - name.find_first_not_of("0123456789", colon + 1) != std::string::npos) { - continue; - } - for (size_t j = 0; j < i; ++j) { - const std::string& first = columns[j].name; - if (base::CaseInsensitiveEqual(first, name.substr(0, colon))) { - return Expected(at, "distinct column names, but there are two named '" + - first + "', which SQLite renamed to '" + first + - "' and '" + name + "'"); - } + // An expression is only named by its alias. + if (name.empty()) { + return Expected(at, "every column to have a name, but column " + + std::to_string(i + 1) + + " has none: give it one with AS"); } - } - for (const NamedColumn& column : columns) { - if (!sql_argument::IsValidColumnName(base::StringView(column.name))) { - return Expected(at, "every column to have a valid name, but '" + - column.name + + if (!sql_argument::IsValidColumnName(base::StringView(name))) { + return Expected(at, "every column to have a valid name, but '" + name + "' is not one: give it one with AS"); } } @@ -422,19 +415,36 @@ op::Scan Compiler::CompileDataframeSource(const dataframe::Dataframe& dataframe, } base::StatusOr Compiler::CompileSqlSource(uint32_t from) { - SqlSource sql = source_(from); - sql = - sql.RewriteAllIgnoreExisting(SqlSource::FromTraceProcessorImplementation( - "SELECT * FROM " + sql.sql())); - auto described = catalog_.DescribeQuery(sql); - if (!described.ok()) { - return base::ErrStatus("%s%s", Traceback(from).c_str(), - described.status().c_message()); + // The columns come from semantic analysis alone: the SQL is only run where + // the pipeline is written, which is the only place everything it reads is + // in scope. + const auto* n = Node(p_, from); + analysis::RelationAnalyzer analyzer(catalog_); + base::StatusOr lineage = + syntaqlite_node_is_present(n->select) + ? analyzer.AnalyzeQuery({p_, n->select}) + : analyzer.AnalyzeRelation(SpanText(p_, n->table_name)); + if (!lineage.ok() || IsPresent(n->schema)) { + std::string reason = + lineage.ok() ? "a schema-qualified table" : lineage.status().message(); + return Err(from, Error::kUnsupported, + "reading a relation whose columns cannot be worked out", + " (" + reason + ")"); } op::Scan scan; - scan.source = std::move(sql); - for (ColumnSchema& column : *described) { - AddScanColumn(scan, std::move(column)); + // The relation as written, which the SQL collecting it reads from. + scan.source = source_(from); + for (const analysis::ColumnLineage& column : lineage->columns()) { + std::optional type; + if (std::optional traced = column.type()) { + type = sql_schema::ToStorageType(*traced); + // An Id's value is the row it sits at, which a query result has no + // rows to point at: it is read as the number it is. + if (type->Is()) { + type = StorageType{core::Uint32{}}; + } + } + AddScanColumn(scan, {std::string(column.output_name), type}); } return scan; } diff --git a/src/trace_processor/perfetto_sql/pipeline/logical_plan.h b/src/trace_processor/perfetto_sql/pipeline/logical_plan.h index efc703f1e0d..79cfe53cadb 100644 --- a/src/trace_processor/perfetto_sql/pipeline/logical_plan.h +++ b/src/trace_processor/perfetto_sql/pipeline/logical_plan.h @@ -67,11 +67,20 @@ struct ScanDataframe { uint32_t row_count = 0; }; +// A relation SQLite collects and hands to the pipeline as its `index`-th +// input. +struct ScanInput { + uint32_t index = 0; +}; + // Reads all rows of a source. Always the first op. struct Scan { using Dataframe = ScanDataframe; - // Where a scan reads from: a dataframe directly, or a query run by SQLite. - using SourceKind = base::TypeSet; + using Input = ScanInput; + // Where a scan reads from: a dataframe directly, or SQL. A plan names SQL by + // its text when compiled; before the plan runs, each SQL source is moved out + // into an input, which SQLite evaluates where the pipeline is written. + using SourceKind = base::TypeSet; internal::VariantOf::type source; // Bindings in source column order. std::vector columns; diff --git a/src/trace_processor/perfetto_sql/pipeline/logical_plan_test_utils.cc b/src/trace_processor/perfetto_sql/pipeline/logical_plan_test_utils.cc index e2072851f15..28fb4853e73 100644 --- a/src/trace_processor/perfetto_sql/pipeline/logical_plan_test_utils.cc +++ b/src/trace_processor/perfetto_sql/pipeline/logical_plan_test_utils.cc @@ -67,6 +67,11 @@ std::string ScanString(const LogicalPlan& plan, const op::Scan& scan) { case Kind::GetTypeIndex(): out += "sql " + base::unchecked_get(scan.source).sql(); break; + case Kind::GetTypeIndex(): + out += "input " + + std::to_string( + base::unchecked_get(scan.source).index); + break; default: PERFETTO_FATAL("Unknown scan source"); } diff --git a/src/trace_processor/perfetto_sql/pipeline/physical_plan.cc b/src/trace_processor/perfetto_sql/pipeline/physical_plan.cc index 5166906e61e..07c277906b3 100644 --- a/src/trace_processor/perfetto_sql/pipeline/physical_plan.cc +++ b/src/trace_processor/perfetto_sql/pipeline/physical_plan.cc @@ -16,6 +16,7 @@ #include "src/trace_processor/perfetto_sql/pipeline/physical_plan.h" +#include #include #include #include @@ -37,7 +38,7 @@ #include "src/trace_processor/core/exec/tree_accumulate.h" #include "src/trace_processor/core/exec/tree_number_nodes.h" #include "src/trace_processor/core/exec/tree_order.h" -#include "src/trace_processor/perfetto_sql/exec/sql_scan.h" +#include "src/trace_processor/perfetto_sql/exec/collected_rows.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" namespace perfetto::trace_processor::pipeline { @@ -132,17 +133,13 @@ std::unique_ptr Lowering::MakeSource(const op::Scan& scan) const { return std::make_unique(source.columns, source.row_count); } - case Kind::GetTypeIndex(): { - Schema columns; - columns.reserve(scan.columns.size()); - for (const NamedColumn& column : scan.columns) { - columns.push_back({column.name, plan_.columns[column.id].type}); - } - return std::make_unique( - env_.connection, base::unchecked_get(scan.source), - std::move(columns), env_.pool); + case Kind::GetTypeIndex(): { + return std::make_unique( + *env_.inputs, + base::unchecked_get(scan.source).index); } default: + // SQL is moved out into inputs before a plan is run. PERFETTO_FATAL("Unknown scan source"); } } @@ -273,10 +270,52 @@ std::unique_ptr Lowering::Finish() { for (const NamedColumn& column : plan_.output) { out_->columns_.push_back({column.name, Position(column.id)}); } + for (const PlanNode& node : plan_.nodes) { + if (!node.Is()) { + continue; + } + const auto& scan = node.Cast(); + if (!std::holds_alternative(scan.source)) { + continue; + } + uint32_t index = base::unchecked_get(scan.source).index; + if (out_->inputs_.size() <= index) { + out_->inputs_.resize(index + 1); + } + for (const NamedColumn& column : scan.columns) { + out_->inputs_[index].push_back( + {column.name, plan_.columns[column.id].type}); + } + } return std::move(out_); } PhysicalPlan::PhysicalPlan() = default; + +base::Status PhysicalPlan::CheckInputs( + const exec::CollectedRowsScan::Inputs& inputs) const { + if (inputs.size() != inputs_.size()) { + return base::ErrStatus("expected %zu inputs, not %zu", inputs_.size(), + inputs.size()); + } + for (size_t i = 0; i < inputs.size(); ++i) { + const exec::CollectedRows& rows = *inputs[i]; + // Rows are described by the first one, so there is nothing to check when + // there are none. + if (rows.batch_count() == 0) { + continue; + } + const core::Schema& columns = rows.columns(); + bool matches = columns.size() == inputs_[i].size(); + for (size_t c = 0; matches && c < columns.size(); ++c) { + matches = columns[c].type == inputs_[i][c].type; + } + if (!matches) { + return base::ErrStatus("input %zu does not have the columns expected", i); + } + } + return base::OkStatus(); +} PhysicalPlan::~PhysicalPlan() = default; std::unique_ptr Lower(const LogicalPlan& plan, diff --git a/src/trace_processor/perfetto_sql/pipeline/physical_plan.h b/src/trace_processor/perfetto_sql/pipeline/physical_plan.h index d9af94566dd..d03dd3d4095 100644 --- a/src/trace_processor/perfetto_sql/pipeline/physical_plan.h +++ b/src/trace_processor/perfetto_sql/pipeline/physical_plan.h @@ -22,22 +22,25 @@ #include #include +#include "perfetto/base/status.h" +#include "src/trace_processor/core/common/schema.h" #include "src/trace_processor/core/exec/operator.h" #include "src/trace_processor/core/exec/pipeline.h" +#include "src/trace_processor/perfetto_sql/exec/collected_rows.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" namespace perfetto::trace_processor { -class SqliteConnection; class StringPool; } // namespace perfetto::trace_processor namespace perfetto::trace_processor::pipeline { -// Connection state needed by lowering. The connection and pool must outlive -// the plan. Dataframe columns have already been resolved by logical planning. +// State needed by lowering, which must outlive the plan. Dataframe columns +// have already been resolved by logical planning. struct LowerEnvironment { - SqliteConnection* connection = nullptr; StringPool* pool = nullptr; + // The rows each run reads for the plan's inputs. + const exec::CollectedRowsScan::Inputs* inputs = nullptr; }; // A pipeline ready to run: the executor nodes plus which batch columns are @@ -58,6 +61,11 @@ class PhysicalPlan { const core::exec::Source& source() const { return *pipeline_; } const std::vector& columns() const { return columns_; } + // Fails unless `inputs` are what a run of the plan reads: one for each of + // its inputs, each with the columns it reads from it. Anyone can write a + // plan and its inputs into SQL, so the two may not have been written + // together. + base::Status CheckInputs(const exec::CollectedRowsScan::Inputs&) const; private: friend class Lowering; @@ -69,6 +77,7 @@ class PhysicalPlan { std::unique_ptr input_; std::unique_ptr pipeline_; std::vector columns_; + std::vector inputs_; }; // Builds executor nodes from a logical plan. Establishes tree numbering, row 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 9b4eba2bd27..0742604e9a8 100644 --- a/src/trace_processor/perfetto_sql/pipeline/physical_plan_unittest.cc +++ b/src/trace_processor/perfetto_sql/pipeline/physical_plan_unittest.cc @@ -30,13 +30,16 @@ #include "perfetto/ext/base/flat_hash_map.h" #include "perfetto/ext/base/status_macros.h" #include "perfetto/ext/base/status_or.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" #include "src/trace_processor/core/exec/column_view.h" #include "src/trace_processor/core/exec/row_cursor.h" #include "src/trace_processor/core/exec/variant.h" +#include "src/trace_processor/perfetto_sql/exec/collected_rows.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/pipeline_sql.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" @@ -83,7 +86,6 @@ class PhysicalPlanTest : public ::testing::Test { PhysicalPlanTest() : connection_(SqliteConnection::CreateConnectionToNewDatabase()), catalog_(&pool_, connection_.get()) { - env_.connection = connection_.get(); env_.pool = &pool_; } @@ -127,7 +129,22 @@ class PhysicalPlanTest : public ::testing::Test { base::StatusOr> Plan(const std::string& sql) { ASSIGN_OR_RETURN(LogicalPlan plan, Compile(sql)); - return Lower(plan, env_); + return LowerWithInputs(std::move(plan)); + } + + // Lowers `plan` to read its SQL sources' rows, collected up front. + base::StatusOr> LowerWithInputs( + LogicalPlan plan) { + RETURN_IF_ERROR(CollectInputs(plan)); + return Lower(MoveSqlSourcesToInputs(std::move(plan)).plan, env_); + } + + // Collects the rows of `plan`'s SQL sources for it to read. + base::Status CollectInputs(const LogicalPlan& plan) { + ASSIGN_OR_RETURN(inputs_, + CollectSqlInputs(connection_.get(), &pool_, plan)); + env_.inputs = &inputs_; + return base::OkStatus(); } std::vector Names(const PhysicalPlan& plan) { @@ -165,6 +182,7 @@ class PhysicalPlanTest : public ::testing::Test { StringPool pool_; std::unique_ptr connection_; TestCatalog catalog_; + exec::CollectedRowsScan::Inputs inputs_; LowerEnvironment env_; base::FlatHashMap macros_; }; @@ -296,7 +314,7 @@ TEST_F(PhysicalPlanTest, OutputBindingsUseIdsRatherThanBatchPositions) { ColumnId id = logical.output.front().id; // Project and alias the same value twice, independently of the source names. logical.output = {{"total", total}, {"id", id}, {"again", total}}; - auto plan = Lower(logical, env_); + auto plan = std::move(*LowerWithInputs(std::move(logical))); EXPECT_THAT(Names(*plan), ElementsAre("total", "id", "again")); EXPECT_NE(plan->columns()[0].index, total); EXPECT_EQ(plan->columns()[0].index, plan->columns()[2].index); @@ -369,7 +387,7 @@ TEST_F(PhysicalPlanTest, DiagnosticsUseDefiningNamesAfterProjection) { // Neither the original input name nor its value is exposed in the result. logical.output = {{"id", logical.output.front().id}, {"renamed_total", logical.output.back().id}}; - auto physical = Lower(logical, env_); + auto physical = std::move(*LowerWithInputs(std::move(logical))); EXPECT_THAT(Run(*physical, "renamed_total").status().message(), HasSubstr("'self'")); } diff --git a/src/trace_processor/perfetto_sql/pipeline/pipeline_sql.cc b/src/trace_processor/perfetto_sql/pipeline/pipeline_sql.cc index 74aebb5e1c4..6f8b56a7d0d 100644 --- a/src/trace_processor/perfetto_sql/pipeline/pipeline_sql.cc +++ b/src/trace_processor/perfetto_sql/pipeline/pipeline_sql.cc @@ -17,30 +17,153 @@ #include "src/trace_processor/perfetto_sql/pipeline/pipeline_sql.h" #include +#include #include +#include +#include +#include #include +#include "perfetto/base/logging.h" #include "perfetto/base/status.h" #include "perfetto/ext/base/status_or.h" #include "perfetto/ext/base/string_utils.h" +#include "perfetto/ext/base/variant.h" +#include "src/trace_processor/core/common/schema.h" +#include "src/trace_processor/core/common/storage_types.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/sqlite/sql_source.h" namespace perfetto::trace_processor::pipeline { +namespace { + +using core::StorageType; + +std::string QuoteIdentifier(const std::string& name) { + return "\"" + base::ReplaceAll(name, "\"", "\"\"") + "\""; +} + +std::string QuoteString(const std::string& text) { + return "'" + base::ReplaceAll(text, "'", "''") + "'"; +} + +// One letter for each type a collected column can hold; `v` carries its type +// per row. +char TypeCode(const std::optional& type) { + if (!type) { + return 'v'; + } + switch (type->index()) { + case StorageType::GetTypeIndex(): + case StorageType::GetTypeIndex(): + // An Id has no storage of its own, so it is collected as a Uint32. + return 'u'; + case StorageType::GetTypeIndex(): + return 'i'; + case StorageType::GetTypeIndex(): + return 'l'; + case StorageType::GetTypeIndex(): + return 'd'; + case StorageType::GetTypeIndex(): + return 's'; + default: + PERFETTO_FATAL("Unknown type"); + } +} + +std::optional> FromTypeCode(char code) { + switch (code) { + case 'v': + return std::optional(); + case 'u': + return StorageType{core::Uint32{}}; + case 'i': + return StorageType{core::Int32{}}; + case 'l': + return StorageType{core::Int64{}}; + case 'd': + return StorageType{core::Double{}}; + case 's': + return StorageType{core::String{}}; + default: + return std::nullopt; + } +} + +} // namespace + +PlanWithInputs MoveSqlSourcesToInputs(LogicalPlan plan) { + PlanWithInputs out; + for (PlanNode& node : plan.nodes) { + if (!node.Is()) { + continue; + } + auto& scan = node.Cast(); + if (!std::holds_alternative(scan.source)) { + continue; + } + core::Schema columns; + std::vector references; + for (const NamedColumn& column : scan.columns) { + columns.push_back({column.name, plan.columns[column.id].type}); + references.push_back(QuoteIdentifier(column.name)); + } + std::string from = base::unchecked_get(scan.source).sql(); + out.inputs.push_back("SELECT " + std::string(kCollectFunction) + "(" + + QuoteString(WriteCollectedColumns(columns)) + ", " + + base::Join(references, ", ") + ") FROM " + from); + scan.source = op::Scan::Input{static_cast(out.inputs.size() - 1)}; + } + out.plan = std::move(plan); + return out; +} base::StatusOr SelectPipelineSql(const LogicalPlan& plan) { if (plan.output.size() > kMaxPipelineColumns) { return base::ErrStatus("A pipeline can output at most %u columns, not %zu", kMaxPipelineColumns, plan.output.size()); } + PlanWithInputs moved = MoveSqlSourcesToInputs(plan); + if (moved.inputs.size() > kMaxPipelineInputs) { + return base::ErrStatus( + "A pipeline can read at most %u relations from SQL, not %zu", + kMaxPipelineInputs, moved.inputs.size()); + } std::vector columns; for (uint32_t i = 0; i < plan.output.size(); ++i) { - columns.push_back("c" + std::to_string(i) + " AS \"" + - base::ReplaceAll(plan.output[i].name, "\"", "\"\"") + - "\""); + columns.push_back("c" + std::to_string(i) + " AS " + + QuoteIdentifier(plan.output[i].name)); + } + std::vector arguments{ + "X'" + base::ToHex(SerializePlan(moved.plan)) + "'"}; + for (const std::string& input : moved.inputs) { + arguments.push_back("(" + input + ")"); } return "SELECT " + base::Join(columns, ", ") + " FROM " + kPipelineFunction + - "(X'" + base::ToHex(SerializePlan(plan)) + "')"; + "(" + base::Join(arguments, ", ") + ")"; +} + +std::string WriteCollectedColumns(const core::Schema& columns) { + std::vector out; + for (const core::ColumnSchema& column : columns) { + out.push_back(std::string(1, TypeCode(column.type)) + ":" + column.name); + } + return base::Join(out, ","); +} + +base::StatusOr ReadCollectedColumns(std::string_view text) { + core::Schema columns; + for (const std::string& part : base::SplitString(std::string(text), ",")) { + std::optional> type = + part.size() >= 2 && part[1] == ':' ? FromTypeCode(part[0]) + : std::nullopt; + if (!type) { + return base::ErrStatus("%s: malformed columns", kCollectFunction); + } + columns.push_back({part.substr(2), *type}); + } + return columns; } } // namespace perfetto::trace_processor::pipeline diff --git a/src/trace_processor/perfetto_sql/pipeline/pipeline_sql.h b/src/trace_processor/perfetto_sql/pipeline/pipeline_sql.h index 4fcc3843131..c0e5baa99d9 100644 --- a/src/trace_processor/perfetto_sql/pipeline/pipeline_sql.h +++ b/src/trace_processor/perfetto_sql/pipeline/pipeline_sql.h @@ -19,8 +19,11 @@ #include #include +#include +#include #include "perfetto/ext/base/status_or.h" +#include "src/trace_processor/core/common/schema.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" namespace perfetto::trace_processor::pipeline { @@ -28,13 +31,33 @@ namespace perfetto::trace_processor::pipeline { // The table function which runs a serialized plan. inline constexpr char kPipelineFunction[] = "__intrinsic_pipeline"; +// The aggregate which collects a relation's rows for a pipeline to read. +inline constexpr char kCollectFunction[] = "__intrinsic_rows"; + // The most columns the table function can output. inline constexpr uint32_t kMaxPipelineColumns = 256; +// The most relations a pipeline can read from SQL. +inline constexpr uint32_t kMaxPipelineInputs = 16; + +// A plan whose SQL sources have been moved out into inputs: the plan reads its +// i-th SQL source as input i, and `inputs[i]` is the SQL collecting it. +struct PlanWithInputs { + LogicalPlan plan; + std::vector inputs; +}; +PlanWithInputs MoveSqlSourcesToInputs(LogicalPlan plan); + // SQL reading `plan`'s output under its own column names. The plan is carried -// in the SQL, serialized, so the SQL runs it wherever it ends up. +// in the SQL, serialized, and each relation it reads from SQL is collected +// where the SQL ends up, so it reads whatever is in scope there: a CTE, or the +// arguments of the function it is written in. base::StatusOr SelectPipelineSql(const LogicalPlan& plan); +// The columns a collection writes: the first argument of kCollectFunction. +std::string WriteCollectedColumns(const core::Schema& columns); +base::StatusOr ReadCollectedColumns(std::string_view); + } // namespace perfetto::trace_processor::pipeline #endif // SRC_TRACE_PROCESSOR_PERFETTO_SQL_PIPELINE_PIPELINE_SQL_H_ diff --git a/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc b/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc index cd78d435924..b36e0d58b3b 100644 --- a/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc +++ b/src/trace_processor/perfetto_sql/pipeline/plan_serialization.cc @@ -16,6 +16,7 @@ #include "src/trace_processor/perfetto_sql/pipeline/plan_serialization.h" +#include #include #include #include @@ -139,10 +140,11 @@ void WriteScan(Writer& w, const op::Scan& scan) { // Its columns are looked up again when read. w.Str(base::unchecked_get(scan.source).name); break; - case Kind::GetTypeIndex(): - w.Str(base::unchecked_get(scan.source).sql()); + case Kind::GetTypeIndex(): + w.U32(base::unchecked_get(scan.source).index); break; default: + // SQL is moved out into inputs before a plan is written. PERFETTO_FATAL("Unknown scan source"); } w.Size(scan.columns.size()); @@ -213,8 +215,8 @@ op::Scan ReadScan(Reader& r) { scan.source = std::move(source); break; } - case Kind::GetTypeIndex(): - scan.source = SqlSource::FromTraceProcessorImplementation(r.Str()); + case Kind::GetTypeIndex(): + scan.source = op::Scan::Input{r.U32()}; break; default: r.Fail(); @@ -294,7 +296,8 @@ base::Status Malformed() { // Checks a plan is one lowering can run as it would a compiled one: a tree // with children before their parents, whose nodes read only columns produced -// below them, each column produced once and with the type it is declared as. +// below them, each column produced once and with the type it is declared as, +// reading its inputs 0 to n - 1 once each. class PlanChecker { public: explicit PlanChecker(const LogicalPlan& plan) @@ -320,6 +323,12 @@ class PlanChecker { return false; } } + std::sort(inputs_.begin(), inputs_.end()); + for (uint32_t i = 0; i < inputs_.size(); ++i) { + if (inputs_[i] != i) { + return false; + } + } return true; } @@ -356,7 +365,11 @@ class PlanChecker { } // Dataframe columns are checked against the dataframe once it is found. // SQLite has no Ids: a query's integers are only ever read as integers. - bool from_sql = std::holds_alternative(scan.source); + bool from_sql = std::holds_alternative(scan.source); + if (from_sql) { + inputs_.push_back( + base::unchecked_get(scan.source).index); + } for (const NamedColumn& column : scan.columns) { if (!Produce(i, column.id) || (from_sql && Is(plan_.columns[column.id].type))) { @@ -449,6 +462,8 @@ class PlanChecker { Columns produced_; std::vector has_parent_; std::vector after_; + // The index of each input a scan reads. + std::vector inputs_; }; // Points each dataframe scan at the dataframe now registered under its name, diff --git a/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc b/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc index 0aab2ef5703..df0cc9438a2 100644 --- a/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc +++ b/src/trace_processor/perfetto_sql/pipeline/plan_serialization_unittest.cc @@ -15,6 +15,7 @@ */ #include "src/trace_processor/perfetto_sql/pipeline/plan_serialization.h" +#include "src/trace_processor/perfetto_sql/pipeline/pipeline_sql.h" #include #include @@ -46,7 +47,6 @@ class PlanSerializationTest : public ::testing::Test { PlanSerializationTest() : connection_(SqliteConnection::CreateConnectionToNewDatabase()), catalog_(&pool_, connection_.get()) { - env_.connection = connection_.get(); env_.pool = &pool_; AddTree(); Exec("CREATE TABLE spans(ts INTEGER, dur INTEGER, cpu INTEGER)"); @@ -69,12 +69,19 @@ class PlanSerializationTest : public ::testing::Test { {{3, 1, 40}, {1, 0, 20}, {2, 0, 30}, {0, std::nullopt, 10}}); } + // The plan as it is written into SQL, with its SQL sources moved out into + // inputs. Their rows are collected for runs of it. 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( + LogicalPlan plan = std::move( std::get(parser.TakeStatement()).plan); + auto inputs = CollectSqlInputs(connection_.get(), &pool_, plan); + PERFETTO_CHECK(inputs.ok()); + inputs_ = std::move(*inputs); + env_.inputs = &inputs_; + return MoveSqlSourcesToInputs(std::move(plan)).plan; } base::StatusOr RoundTrip(const LogicalPlan& plan) { @@ -84,6 +91,7 @@ class PlanSerializationTest : public ::testing::Test { StringPool pool_; std::unique_ptr connection_; TestCatalog catalog_; + exec::CollectedRowsScan::Inputs inputs_; LowerEnvironment env_; base::FlatHashMap macros_; }; @@ -140,6 +148,9 @@ TEST_F(PlanSerializationTest, CorruptPlansAreRefusedOrRun) { } // Whatever passes the checks must be safe to lower and run. auto physical = Lower(*read, env_); + if (!physical->CheckInputs(inputs_).ok()) { + continue; + } core::exec::RowCursor cursor(physical->source()); for (bool row = cursor.Open(); row; row = cursor.Next()) { } diff --git a/src/trace_processor/perfetto_sql/pipeline/test_catalog.h b/src/trace_processor/perfetto_sql/pipeline/test_catalog.h index 10383d20f49..798ae03a0cd 100644 --- a/src/trace_processor/perfetto_sql/pipeline/test_catalog.h +++ b/src/trace_processor/perfetto_sql/pipeline/test_catalog.h @@ -28,22 +28,26 @@ #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 "perfetto/ext/base/string_utils.h" #include "src/perfetto_sql/analysis/relation.h" #include "src/trace_processor/containers/string_pool.h" #include "src/trace_processor/core/dataframe/adhoc_dataframe_builder.h" #include "src/trace_processor/core/dataframe/dataframe.h" +#include "src/trace_processor/perfetto_sql/exec/collected_rows.h" #include "src/trace_processor/perfetto_sql/pipeline/catalog.h" #include "src/trace_processor/perfetto_sql/pipeline/logical_plan.h" -#include "src/trace_processor/perfetto_sql/schema/query_schema.h" +#include "src/trace_processor/perfetto_sql/schema/sqlite_relations.h" +#include "src/trace_processor/perfetto_sql/schema/type_mapping.h" #include "src/trace_processor/sqlite/sql_source.h" #include "src/trace_processor/sqlite/sqlite_connection.h" namespace perfetto::trace_processor::pipeline { // Catalog over dataframes built by the test. SQLite does not know about them. -// Anything else is described via `connection` (if given) and is untyped. -class TestCatalog : public Catalog, public perfetto_sql::analysis::Catalog { +// Anything else is looked up in `connection` (if given) and is untyped. +class TestCatalog : public Catalog { public: explicit TestCatalog(StringPool* pool, SqliteConnection* connection = nullptr) : pool_(pool), connection_(connection) {} @@ -80,16 +84,22 @@ class TestCatalog : public Catalog, public perfetto_sql::analysis::Catalog { return dataframe ? dataframe->get() : nullptr; } - base::StatusOr DescribeQuery(const SqlSource& sql) const override { - if (!connection_) { - return base::ErrStatus("no such table"); - } - return sql_schema::DescribeQuery(connection_, sql, *this); - } - std::optional FindLeafRelation( - std::string_view) const override { - return std::nullopt; + std::string_view name) const override { + const dataframe::Dataframe* dataframe = FindDataframe(name); + if (!dataframe) { + return connection_ ? sql_schema::FindSqliteRelation(connection_, name) + : std::nullopt; + } + perfetto_sql::analysis::LeafRelation relation; + relation.name = std::string(name); + const std::vector& columns = dataframe->column_names(); + for (uint32_t i = 0; i < columns.size(); ++i) { + relation.columns.push_back( + {columns[i], sql_schema::ToAnalysisType(dataframe->column_type(i)), + dataframe::IsHiddenColumn(columns[i])}); + } + return relation; } std::optional FindViewSql(std::string_view) const override { return std::nullopt; @@ -102,6 +112,46 @@ class TestCatalog : public Catalog, public perfetto_sql::analysis::Catalog { dataframes_; }; +// Collects the rows of each of `plan`'s SQL sources from `connection`, as +// SQLite does where the pipeline is written: the inputs a plan with its SQL +// sources moved out reads. +inline base::StatusOr CollectSqlInputs( + SqliteConnection* connection, + StringPool* pool, + const LogicalPlan& plan) { + exec::CollectedRowsScan::Inputs inputs; + for (const PlanNode& node : plan.nodes) { + if (!node.Is()) { + continue; + } + const auto& scan = node.Cast(); + if (!std::holds_alternative(scan.source)) { + continue; + } + core::Schema columns; + std::vector names; + for (const NamedColumn& column : scan.columns) { + columns.push_back({column.name, plan.columns[column.id].type}); + names.push_back("\"" + column.name + "\""); + } + auto rows = std::make_shared(columns, pool); + auto statement = connection->PrepareStatement(SqlSource::FromExecuteQuery( + "SELECT " + base::Join(names, ", ") + " FROM " + + base::unchecked_get(scan.source).sql())); + while (statement.Step()) { + std::vector values; + for (uint32_t i = 0; i < columns.size(); ++i) { + values.push_back( + sqlite3_column_value(statement.sqlite_stmt(), static_cast(i))); + } + RETURN_IF_ERROR(rows->Append(values.data())); + } + RETURN_IF_ERROR(statement.status()); + inputs.push_back(std::move(rows)); + } + return inputs; +} + } // namespace perfetto::trace_processor::pipeline #endif // SRC_TRACE_PROCESSOR_PERFETTO_SQL_PIPELINE_TEST_CATALOG_H_ diff --git a/src/trace_processor/perfetto_sql/schema/BUILD.gn b/src/trace_processor/perfetto_sql/schema/BUILD.gn index ec7dd41e623..327dcf7a718 100644 --- a/src/trace_processor/perfetto_sql/schema/BUILD.gn +++ b/src/trace_processor/perfetto_sql/schema/BUILD.gn @@ -14,8 +14,8 @@ source_set("schema") { sources = [ - "query_schema.cc", - "query_schema.h", + "sqlite_relations.cc", + "sqlite_relations.h", "type_mapping.h", ] deps = [ diff --git a/src/trace_processor/perfetto_sql/schema/query_schema.cc b/src/trace_processor/perfetto_sql/schema/query_schema.cc deleted file mode 100644 index 70767614184..00000000000 --- a/src/trace_processor/perfetto_sql/schema/query_schema.cc +++ /dev/null @@ -1,112 +0,0 @@ -/* - * 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/schema/query_schema.h" - -#include - -#include -#include -#include -#include - -#include "perfetto/base/status.h" -#include "perfetto/ext/base/status_macros.h" -#include "src/perfetto_sql/syntaqlite/syntaqlite_perfetto.h" -#include "src/trace_processor/perfetto_sql/schema/type_mapping.h" -#include "src/trace_processor/sqlite/bindings/sqlite_column.h" - -namespace perfetto::trace_processor::sql_schema { -namespace { - -using core::StorageType; - -struct ParserDeleter { - void operator()(SyntaqliteParser* parser) const { - syntaqlite_parser_destroy(parser); - } -}; -using ScopedParser = std::unique_ptr; - -// The types lineage established, lined up with the query's columns. If the two -// disagree on the number of columns they are not describing the same query, so -// no type is claimed for any of them. -std::vector> ResolveTypes( - const SqlSource& sql, - uint32_t count, - const analysis::Catalog& catalog) { - std::vector> types(count); - ScopedParser parser(syntaqlite_parser_create_perfetto(nullptr)); - syntaqlite_parser_reset(parser.get(), sql.sql().data(), - static_cast(sql.sql().size())); - if (syntaqlite_parser_next(parser.get()) != SYNTAQLITE_PARSE_OK) { - return types; - } - analysis::RelationAnalyzer analyzer(catalog); - auto resolved = analyzer.AnalyzeQuery( - {parser.get(), syntaqlite_result_root(parser.get())}); - if (!resolved.ok() || resolved->columns().size() != count) { - return types; - } - for (uint32_t i = 0; i < count; ++i) { - std::optional type = resolved->columns()[i].type(); - if (!type) { - continue; - } - StorageType storage = ToStorageType(*type); - // An Id has no storage of its own: its value is the row it sits at. A - // query result has no such rows to point at, so materialise it at the - // narrowest width which holds one. - if (storage.Is()) { - storage = StorageType{core::Uint32{}}; - } - types[i] = storage; - } - return types; -} - -} // namespace - -base::StatusOr DescribeQuery(SqliteConnection* connection, - const SqlSource& sql, - const analysis::Catalog& catalog) { - // Prepared here only to read the column names, then discarded: a statement - // belongs to one execution, but the columns belong to the query. - SqliteConnection::PreparedStatement statement = - connection->PrepareStatement(sql); - RETURN_IF_ERROR(statement.status()); - - sqlite3_stmt* stmt = statement.sqlite_stmt(); - // Such as a function's argument, when the query is in a function's body. - if (sqlite3_bind_parameter_count(stmt) > 0) { - const char* name = sqlite3_bind_parameter_name(stmt, 1); - return base::ErrStatus( - "Cannot read `%s`: a pipeline cannot read function arguments yet", - name ? name : "?"); - } - uint32_t count = sqlite::column::Count(stmt); - std::vector> types = - ResolveTypes(sql, count, catalog); - core::Schema columns; - columns.reserve(count); - for (uint32_t i = 0; i < count; ++i) { - const char* name = sqlite::column::Name(stmt, i); - columns.push_back({name ? name : "", types[i]}); - } - return columns; -} - -} // namespace perfetto::trace_processor::sql_schema diff --git a/src/trace_processor/perfetto_sql/schema/sqlite_relations.cc b/src/trace_processor/perfetto_sql/schema/sqlite_relations.cc new file mode 100644 index 00000000000..79658e43384 --- /dev/null +++ b/src/trace_processor/perfetto_sql/schema/sqlite_relations.cc @@ -0,0 +1,53 @@ +/* + * 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/schema/sqlite_relations.h" + +#include +#include +#include + +#include "perfetto/ext/base/string_utils.h" +#include "src/perfetto_sql/analysis/relation.h" +#include "src/trace_processor/sqlite/bindings/sqlite_column.h" +#include "src/trace_processor/sqlite/sql_source.h" +#include "src/trace_processor/sqlite/sqlite_connection.h" + +namespace perfetto::trace_processor::sql_schema { + +std::optional FindSqliteRelation( + SqliteConnection* connection, + std::string_view name) { + std::string quoted = + "'" + base::ReplaceAll(std::string(name), "'", "''") + "'"; + SqliteConnection::PreparedStatement stmt = + connection->PrepareStatement(SqlSource::FromTraceProcessorImplementation( + "SELECT name, hidden FROM pragma_table_xinfo(" + quoted + ")")); + perfetto_sql::analysis::LeafRelation relation; + relation.name = std::string(name); + while (stmt.Step()) { + const char* column = sqlite::column::Text(stmt.sqlite_stmt(), 0); + // 1 is a hidden column; 2 and 3 are generated ones, which `*` includes. + bool hidden = sqlite::column::Int64(stmt.sqlite_stmt(), 1) == 1; + relation.columns.push_back({column ? column : "", std::nullopt, hidden}); + } + if (!stmt.status().ok() || relation.columns.empty()) { + return std::nullopt; + } + return relation; +} + +} // namespace perfetto::trace_processor::sql_schema diff --git a/src/trace_processor/perfetto_sql/schema/query_schema.h b/src/trace_processor/perfetto_sql/schema/sqlite_relations.h similarity index 55% rename from src/trace_processor/perfetto_sql/schema/query_schema.h rename to src/trace_processor/perfetto_sql/schema/sqlite_relations.h index 5fbf25542b1..cb0f6a47ff7 100644 --- a/src/trace_processor/perfetto_sql/schema/query_schema.h +++ b/src/trace_processor/perfetto_sql/schema/sqlite_relations.h @@ -14,26 +14,25 @@ * limitations under the License. */ -#ifndef SRC_TRACE_PROCESSOR_PERFETTO_SQL_SCHEMA_QUERY_SCHEMA_H_ -#define SRC_TRACE_PROCESSOR_PERFETTO_SQL_SCHEMA_QUERY_SCHEMA_H_ +#ifndef SRC_TRACE_PROCESSOR_PERFETTO_SQL_SCHEMA_SQLITE_RELATIONS_H_ +#define SRC_TRACE_PROCESSOR_PERFETTO_SQL_SCHEMA_SQLITE_RELATIONS_H_ + +#include +#include -#include "perfetto/ext/base/status_or.h" #include "src/perfetto_sql/analysis/relation.h" -#include "src/trace_processor/core/common/schema.h" -#include "src/trace_processor/sqlite/sql_source.h" #include "src/trace_processor/sqlite/sqlite_connection.h" namespace perfetto::trace_processor::sql_schema { -// SQLite supplies result names; semantic analysis supplies types where known. -// Fails if SQLite cannot prepare the query, or if it reads a parameter: the -// query is run on its own, where nothing binds one. Unknown types remain -// per-row variants. -base::StatusOr DescribeQuery( +// The columns of the table, virtual table or table function SQLite knows as +// `name`, or nothing when it knows none. They are untyped: SQLite's declared +// types establish nothing, as an INTEGER column holds text if something puts +// text in it. +std::optional FindSqliteRelation( SqliteConnection*, - const SqlSource&, - const perfetto_sql::analysis::Catalog&); + std::string_view name); } // namespace perfetto::trace_processor::sql_schema -#endif // SRC_TRACE_PROCESSOR_PERFETTO_SQL_SCHEMA_QUERY_SCHEMA_H_ +#endif // SRC_TRACE_PROCESSOR_PERFETTO_SQL_SCHEMA_SQLITE_RELATIONS_H_