tp: read a pipeline's rows from a query

A pipeline of operators has had nowhere to get its rows from except an
array built by hand, which is enough to test an operator and not enough
to run one over a trace. Everything a query would want to feed it is
already SQL, and stays SQL for the foreseeable future, so the way in has
to be a query.

This is that way in: a source which steps a prepared statement and fills
batches from it. Nothing downstream learns that SQLite produced the
rows, and nothing downstream has to be taught to. It is also the first
source whose storage does not stand still, which is what holding a
batch's values rather than a view of them was for.

The column types are inferred, which is a stopgap: a bare query is all
this is given, so it settles each column on its first non-null value and
refuses a later value which disagrees, because by then the earlier rows
have gone downstream and cannot be retyped. Once a FROM stage is built
from a logical plan it will be handed the types and do as it is told.

Author: LalitMaganti

Android.bp

@@ -18771,6 +18771,22 @@ filegroup {
     ],
 }
 
+// GN: //src/trace_processor/perfetto_sql/exec:exec
+filegroup {
+    name: "perfetto_src_trace_processor_perfetto_sql_exec_exec",
+    srcs: [
+        "src/trace_processor/perfetto_sql/exec/sql_scan.cc",
+    ],
+}
+
+// GN: //src/trace_processor/perfetto_sql/exec:unittests
+filegroup {
+    name: "perfetto_src_trace_processor_perfetto_sql_exec_unittests",
+    srcs: [
+        "src/trace_processor/perfetto_sql/exec/sql_scan_unittest.cc",
+    ],
+}
+
 // GN: //src/trace_processor/perfetto_sql/generator:gen_cc_perfetto_sql_descriptor
 genrule {
     name: "perfetto_src_trace_processor_perfetto_sql_generator_gen_cc_perfetto_sql_descriptor",
@@ -24050,6 +24066,8 @@ cc_test {
         ":perfetto_src_trace_processor_metrics_unittests",
         ":perfetto_src_trace_processor_perfetto_sql_engine_engine",
         ":perfetto_src_trace_processor_perfetto_sql_engine_unittests",
+        ":perfetto_src_trace_processor_perfetto_sql_exec_exec",
+        ":perfetto_src_trace_processor_perfetto_sql_exec_unittests",
         ":perfetto_src_trace_processor_perfetto_sql_generator_generator",
         ":perfetto_src_trace_processor_perfetto_sql_generator_unittests",
         ":perfetto_src_trace_processor_perfetto_sql_intrinsics_types_types",

src/trace_processor/BUILD.gn

@@ -469,6 +469,7 @@ perfetto_unittest_source_set("unittests") {
   if (enable_perfetto_trace_processor_sqlite) {
     deps += [
       "perfetto_sql/engine:unittests",
+      "perfetto_sql/exec:unittests",
       "perfetto_sql/parser:unittests",
       "perfetto_sql/tokenizer:unittests",
       "plugins/ancestor:unittests",

src/trace_processor/perfetto_sql/exec/BUILD.gn

@@ -0,0 +1,47 @@
+# 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.
+
+import("../../../../gn/test.gni")
+
+source_set("exec") {
+  sources = [
+    "sql_scan.cc",
+    "sql_scan.h",
+  ]
+  deps = [
+    "../../../../gn:default_deps",
+    "../../../../gn:sqlite",
+    "../../../base",
+    "../../containers",
+    "../../core/common",
+    "../../core/exec",
+    "../../core/util",
+    "../../sqlite",
+  ]
+}
+
+perfetto_unittest_source_set("unittests") {
+  testonly = true
+  sources = [ "sql_scan_unittest.cc" ]
+  deps = [
+    ":exec",
+    "../../../../gn:default_deps",
+    "../../../../gn:gtest_and_gmock",
+    "../../../base",
+    "../../containers",
+    "../../core/common",
+    "../../core/exec",
+    "../../sqlite",
+  ]
+}

src/trace_processor/perfetto_sql/exec/sql_scan.cc

@@ -0,0 +1,224 @@
+/*
+ * 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 <sqlite3.h>
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "perfetto/base/status.h"
+#include "perfetto/ext/base/status_macros.h"
+#include "perfetto/ext/base/status_or.h"
+#include "src/trace_processor/containers/string_pool.h"
+#include "src/trace_processor/core/common/storage_types.h"
+#include "src/trace_processor/core/exec/column_view.h"
+#include "src/trace_processor/core/exec/row_batch.h"
+#include "src/trace_processor/core/exec/row_selection.h"
+#include "src/trace_processor/core/util/bit_vector.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::ColumnView;
+using core::exec::kMaxBatchRows;
+using core::exec::RowBatch;
+using core::exec::RowSelection;
+
+const char* TypeName(StorageType type) {
+  if (type.Is<Int64>()) {
+    return "an integer";
+  }
+  if (type.Is<Double>()) {
+    return "a float";
+  }
+  return "a string";
+}
+
+}  // namespace
+
+base::StatusOr<std::unique_ptr<SqlScan>>
+SqlScan::Create(SqliteConnection* connection, SqlSource sql, StringPool* pool) {
+  SqliteConnection::PreparedStatement statement =
+      connection->PrepareStatement(std::move(sql));
+  RETURN_IF_ERROR(statement.status());
+
+  // The columns are known once the statement is prepared, before any row is
+  // read, which is what lets a stage resolve the names it was given first.
+  sqlite3_stmt* stmt = statement.sqlite_stmt();
+  auto count = static_cast<uint32_t>(sqlite3_column_count(stmt));
+  std::vector<std::string> names;
+  names.reserve(count);
+  for (uint32_t i = 0; i < count; ++i) {
+    const char* name = sqlite3_column_name(stmt, static_cast<int>(i));
+    names.emplace_back(name ? name : "");
+  }
+  return std::unique_ptr<SqlScan>(
+      new SqlScan(std::move(statement), std::move(names), pool));
+}
+
+SqlScan::SqlScan(SqliteConnection::PreparedStatement statement,
+                 std::vector<std::string> names,
+                 StringPool* pool)
+    : statement_(std::move(statement)),
+      names_(std::move(names)),
+      pool_(pool),
+      columns_(names_.size()) {
+  for (Column& column : columns_) {
+    column.validity = core::BitVector::CreateWithSize(kMaxBatchRows);
+  }
+}
+
+SqlScan::~SqlScan() = default;
+
+void SqlScan::Reset() {
+  sqlite3_reset(statement_.sqlite_stmt());
+  done_ = false;
+}
+
+bool SqlScan::Settle(uint32_t index, StorageType type) {
+  Column& column = columns_[index];
+  if (column.type) {
+    if (*column.type == type) {
+      return true;
+    }
+    status_ = base::ErrStatus(
+        "SQL source: column '%s' holds %s after already holding %s; a "
+        "column's type is settled by its first value and cannot change",
+        names_[index].c_str(), TypeName(type), TypeName(*column.type));
+    return false;
+  }
+  column.type = type;
+  if (type.Is<Int64>()) {
+    column.ints.resize(kMaxBatchRows);
+  } else if (type.Is<Double>()) {
+    column.doubles.resize(kMaxBatchRows);
+  } else {
+    column.strings.resize(kMaxBatchRows);
+  }
+  return true;
+}
+
+bool SqlScan::ReadValue(sqlite3_stmt* stmt, uint32_t index, uint32_t row) {
+  Column& column = columns_[index];
+  auto col = static_cast<int>(index);
+  switch (sqlite3_column_type(stmt, col)) {
+    case SQLITE_NULL:
+      // The validity bit was cleared when the batch started.
+      return true;
+    case SQLITE_INTEGER:
+      if (!Settle(index, StorageType{Int64{}})) {
+        return false;
+      }
+      column.ints[row] = sqlite3_column_int64(stmt, col);
+      break;
+    case SQLITE_FLOAT:
+      if (!Settle(index, StorageType{Double{}})) {
+        return false;
+      }
+      column.doubles[row] = sqlite3_column_double(stmt, col);
+      break;
+    case SQLITE_TEXT:
+      if (!Settle(index, StorageType{String{}})) {
+        return false;
+      }
+      column.strings[row] = pool_->InternString(
+          reinterpret_cast<const char*>(sqlite3_column_text(stmt, col)));
+      break;
+    default:
+      status_ = base::ErrStatus(
+          "SQL source: column '%s' holds a blob, which a pipeline cannot "
+          "carry",
+          names_[index].c_str());
+      return false;
+  }
+  column.validity.set(row);
+  return true;
+}
+
+ColumnView SqlScan::MakeView(const Column& column) const {
+  const void* data = nullptr;
+  if (column.type->Is<Int64>()) {
+    data = column.ints.data();
+  } else if (column.type->Is<Double>()) {
+    data = column.doubles.data();
+  } else {
+    data = column.strings.data();
+  }
+  return ColumnView::Reference(*column.type, data, &column.validity);
+}
+
+RowBatch* SqlScan::Next() {
+  if (done_ || !status_.ok()) {
+    return nullptr;
+  }
+  for (Column& column : columns_) {
+    column.validity.ClearAllBits();
+  }
+
+  sqlite3_stmt* stmt = statement_.sqlite_stmt();
+  uint32_t count = 0;
+  while (count < kMaxBatchRows && statement_.Step()) {
+    for (uint32_t i = 0; i < columns_.size(); ++i) {
+      if (!ReadValue(stmt, i, count)) {
+        return nullptr;
+      }
+    }
+    ++count;
+  }
+  if (!statement_.status().ok()) {
+    status_ = statement_.status();
+    return nullptr;
+  }
+  done_ = count < kMaxBatchRows;
+  if (count == 0) {
+    return nullptr;
+  }
+
+  if (batch_.column_count() == 0) {
+    for (uint32_t i = 0; i < columns_.size(); ++i) {
+      // A column which never held a value still has to be some type for
+      // anything to read it; nothing will, because none of its rows are valid.
+      if (!columns_[i].type) {
+        Settle(i, StorageType{Int64{}});
+      }
+      pristine_.push_back(MakeView(columns_[i]));
+      batch_.AddColumn(pristine_.back());
+    }
+  }
+
+  batch_.PrepareForFill();
+  for (uint32_t i = 0; i < columns_.size(); ++i) {
+    // Operators only replace a column's row view, so restoring the views is
+    // all a reused batch needs before being filled again.
+    batch_.mutable_column(i).AdoptSelection(pristine_[i]);
+  }
+  batch_.Compose(RowSelection::Range(0), count);
+  batch_.SetCardinality(count);
+  return &batch_;
+}
+
+}  // namespace perfetto::trace_processor::exec