diff --git a/Android.bp b/Android.bp index 705dd53812..a967f781db 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 fa2209888f..b21b48dfa4 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 29477fa924..14f186a3cb 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 c832464a9a..c0acbbf578 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 8cf1357d0d..7608b71d06 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 7221007cf1..f573e03493 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 cfa7433413..079a742c6a 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 895222dc9b..a682fdf3d7 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 696374b910..f61530f000 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 46d709da8f..825874f93c 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 0caaead96a..306013a0fa 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 d559857494..272b91ea0a 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 0000000000..7506cda517 --- /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 0000000000..8007c417ed --- /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 0000000000..f5e03b0a67 --- /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 7f1bef0657..0000000000 --- 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 1cd11d9575..0000000000 --- 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 56693cd587..0000000000 --- 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 2a855572a3..7932e863db 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 02ed414507..89a513956a 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 2c4c874f5b..c23391c82e 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 7b169d55c3..bc290babc7 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 8c38141751..a128a68b78 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 6e62e52325..6a6e41dd96 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 afff675090..3cd6f1fde4 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 efc703f1e0..79cfe53cad 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 e2072851f1..28fb4853e7 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 5166906e61..07c277906b 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 d9af94566d..d03dd3d409 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 9b4eba2bd2..0742604e9a 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 74aebb5e1c..6f8b56a7d0 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 4fcc384313..c0e5baa99d 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 cd78d43592..b36e0d58b3 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 0aab2ef570..df0cc9438a 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 10383d20f4..798ae03a0c 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 ec7dd41e62..327dcf7a71 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 7076761418..0000000000 --- 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 0000000000..79658e4338 --- /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 5fbf25542b..cb0f6a47ff 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_