diff --git a/Android.bp b/Android.bp
index 48a0cfa..77ddad6 100644
--- a/Android.bp
+++ b/Android.bp
@@ -10428,6 +10428,7 @@
         "src/trace_processor/sqlite/db_sqlite_table.cc",
         "src/trace_processor/sqlite/perfetto_sql_engine.cc",
         "src/trace_processor/sqlite/perfetto_sql_parser.cc",
+        "src/trace_processor/sqlite/sql_source.cc",
         "src/trace_processor/sqlite/sql_stats_table.cc",
         "src/trace_processor/sqlite/sqlite_engine.cc",
         "src/trace_processor/sqlite/sqlite_table.cc",
diff --git a/BUILD b/BUILD
index d4ce249..4cb49b9 100644
--- a/BUILD
+++ b/BUILD
@@ -2208,6 +2208,8 @@
         "src/trace_processor/sqlite/perfetto_sql_parser.h",
         "src/trace_processor/sqlite/query_cache.h",
         "src/trace_processor/sqlite/scoped_db.h",
+        "src/trace_processor/sqlite/sql_source.cc",
+        "src/trace_processor/sqlite/sql_source.h",
         "src/trace_processor/sqlite/sql_stats_table.cc",
         "src/trace_processor/sqlite/sql_stats_table.h",
         "src/trace_processor/sqlite/sqlite_engine.cc",
diff --git a/include/perfetto/base/status.h b/include/perfetto/base/status.h
index f059184..506bb6e 100644
--- a/include/perfetto/base/status.h
+++ b/include/perfetto/base/status.h
@@ -80,7 +80,7 @@
   // Gets the payload for the given |type_url| if one exists.
   //
   // Will always return std::nullopt if |ok()|.
-  std::optional<std::string_view> GetPayload(std::string_view type_url);
+  std::optional<std::string_view> GetPayload(std::string_view type_url) const;
 
   // Sets the payload for the given key. The key should
   //
diff --git a/include/perfetto/trace_processor/basic_types.h b/include/perfetto/trace_processor/basic_types.h
index 9726bbb..0bbfcfa 100644
--- a/include/perfetto/trace_processor/basic_types.h
+++ b/include/perfetto/trace_processor/basic_types.h
@@ -254,7 +254,7 @@
 
   // If true, SqlModule will override registered module with the same name. Can
   // only be set if enable_dev_features is true, otherwise will throw an error.
-  bool allow_module_override;
+  bool allow_module_override = false;
 };
 
 }  // namespace trace_processor
diff --git a/src/base/status.cc b/src/base/status.cc
index d3c13e8..58e32d1 100644
--- a/src/base/status.cc
+++ b/src/base/status.cc
@@ -32,7 +32,8 @@
   return status;
 }
 
-std::optional<std::string_view> Status::GetPayload(std::string_view type_url) {
+std::optional<std::string_view> Status::GetPayload(
+    std::string_view type_url) const {
   if (ok()) {
     return std::nullopt;
   }
diff --git a/src/trace_processor/iterator_impl.h b/src/trace_processor/iterator_impl.h
index f9a824a..722c8c6 100644
--- a/src/trace_processor/iterator_impl.h
+++ b/src/trace_processor/iterator_impl.h
@@ -25,11 +25,14 @@
 
 #include "perfetto/base/build_config.h"
 #include "perfetto/base/export.h"
+#include "perfetto/base/status.h"
+#include "perfetto/ext/base/status_or.h"
 #include "perfetto/trace_processor/basic_types.h"
 #include "perfetto/trace_processor/iterator.h"
 #include "perfetto/trace_processor/status.h"
 #include "src/trace_processor/sqlite/perfetto_sql_engine.h"
 #include "src/trace_processor/sqlite/scoped_db.h"
+#include "src/trace_processor/sqlite/sqlite_engine.h"
 #include "src/trace_processor/sqlite/sqlite_utils.h"
 
 namespace perfetto {
@@ -52,65 +55,61 @@
 
   // Methods called by the base Iterator class.
   bool Next() {
+    // In the past, we used to call sqlite3_step for the first time in this
+    // function which 1:1 matched Next calls to sqlite3_step calls. However,
+    // with the introduction of multi-statement support, we tokenize the
+    // queries and so we need to *not* call step the first time Next is
+    // called.
+    //
+    // Aside: if we could, we would change the API to match the new setup
+    // (i.e. implement operator bool, make Next return nothing similar to C++
+    // iterators); however, too many clients depend on the current behavior so
+    // we have to keep the API as is.
     if (!called_next_) {
       // Delegate to the cc file to prevent trace_storage.h include in this
       // file.
       RecordFirstNextInSqlStats();
       called_next_ = true;
-
-      // In the past, we used to call sqlite3_step for the first time in this
-      // function which 1:1 matched Next calls to sqlite3_step calls. However,
-      // with the introduction of multi-statement support, we call
-      // sqlite3_step when tokenizing the queries and so we need to *not* call
-      // step the first time Next is called.
-      //
-      // Aside: if we could, we would change the API to match the new setup
-      // (i.e. implement operator bool, make Next return nothing similar to C++
-      // iterators); however, too many clients depend on the current behavior so
-      // we have to keep the API as is.
-      return result_.ok() && !sqlite_utils::IsStmtDone(*result_->stmt);
+      return result_.ok() && !result_->stmt.IsDone();
     }
-
-    if (!result_.ok())
-      return false;
-
-    int ret = sqlite3_step(*result_->stmt);
-    if (PERFETTO_UNLIKELY(ret != SQLITE_ROW && ret != SQLITE_DONE)) {
-      result_ =
-          base::ErrStatus("%s", sqlite_utils::FormatErrorMessage(
-                                    result_->stmt.get(), std::nullopt,
-                                    sqlite3_db_handle(result_->stmt.get()), ret)
-                                    .c_message());
+    if (!result_.ok()) {
       return false;
     }
-    return ret == SQLITE_ROW;
+
+    bool has_more = result_->stmt.Step();
+    if (!result_->stmt.status().ok()) {
+      PERFETTO_DCHECK(!has_more);
+      result_ = result_->stmt.status();
+    }
+    return has_more;
   }
 
   SqlValue Get(uint32_t col) const {
     PERFETTO_DCHECK(result_.ok());
 
     auto column = static_cast<int>(col);
-    auto col_type = sqlite3_column_type(*result_->stmt, column);
+    sqlite3_stmt* stmt = result_->stmt.sqlite_stmt();
+    auto col_type = sqlite3_column_type(stmt, column);
     SqlValue value;
     switch (col_type) {
       case SQLITE_INTEGER:
         value.type = SqlValue::kLong;
-        value.long_value = sqlite3_column_int64(*result_->stmt, column);
+        value.long_value = sqlite3_column_int64(stmt, column);
         break;
       case SQLITE_TEXT:
         value.type = SqlValue::kString;
-        value.string_value = reinterpret_cast<const char*>(
-            sqlite3_column_text(*result_->stmt, column));
+        value.string_value =
+            reinterpret_cast<const char*>(sqlite3_column_text(stmt, column));
         break;
       case SQLITE_FLOAT:
         value.type = SqlValue::kDouble;
-        value.double_value = sqlite3_column_double(*result_->stmt, column);
+        value.double_value = sqlite3_column_double(stmt, column);
         break;
       case SQLITE_BLOB:
         value.type = SqlValue::kBytes;
-        value.bytes_value = sqlite3_column_blob(*result_->stmt, column);
+        value.bytes_value = sqlite3_column_blob(stmt, column);
         value.bytes_count =
-            static_cast<size_t>(sqlite3_column_bytes(*result_->stmt, column));
+            static_cast<size_t>(sqlite3_column_bytes(stmt, column));
         break;
       case SQLITE_NULL:
         value.type = SqlValue::kNull;
@@ -120,27 +119,27 @@
   }
 
   std::string GetColumnName(uint32_t col) const {
-    return result_.ok() && result_->stmt
-               ? sqlite3_column_name(*result_->stmt, static_cast<int>(col))
-               : "";
+    return result_.ok() ? sqlite3_column_name(result_->stmt.sqlite_stmt(),
+                                              static_cast<int>(col))
+                        : "";
   }
 
   base::Status Status() const { return result_.status(); }
 
   uint32_t ColumnCount() const {
-    return result_.ok() ? result_->column_count : 0;
+    return result_.ok() ? result_->stats.column_count : 0;
   }
 
   uint32_t StatementCount() const {
-    return result_.ok() ? result_->statement_count : 0;
+    return result_.ok() ? result_->stats.statement_count : 0;
   }
 
   uint32_t StatementCountWithOutput() const {
-    return result_.ok() ? result_->statement_count_with_output : 0;
+    return result_.ok() ? result_->stats.statement_count_with_output : 0;
   }
 
   std::string LastStatementSql() const {
-    return result_.ok() ? sqlite3_sql(*result_->stmt) : "";
+    return result_.ok() ? result_->stmt.sql() : "";
   }
 
  private:
@@ -160,7 +159,6 @@
 
   ScopedTraceProcessor trace_processor_;
   base::StatusOr<PerfettoSqlEngine::ExecutionResult> result_;
-
   uint32_t sql_stats_row_ = 0;
   bool called_next_ = false;
 };
diff --git a/src/trace_processor/metrics/metrics.cc b/src/trace_processor/metrics/metrics.cc
index 34c7863..5b0e727 100644
--- a/src/trace_processor/metrics/metrics.cc
+++ b/src/trace_processor/metrics/metrics.cc
@@ -24,6 +24,9 @@
 #include "perfetto/ext/base/string_utils.h"
 #include "perfetto/ext/base/utils.h"
 #include "perfetto/protozero/scattered_heap_buffer.h"
+#include "sqlite3.h"
+#include "src/trace_processor/sqlite/perfetto_sql_engine.h"
+#include "src/trace_processor/sqlite/sql_source.h"
 #include "src/trace_processor/sqlite/sqlite_utils.h"
 #include "src/trace_processor/tp_metatrace.h"
 #include "src/trace_processor/util/descriptors.h"
@@ -651,15 +654,8 @@
         metric_it->sql.c_str());
   }
 
-  auto it = ctx->tp->ExecuteQuery(subbed_sql);
-  it.Next();
-
-  base::Status status = it.Status();
-  if (!status.ok()) {
-    return base::ErrStatus("RUN_METRIC: Error when running file %s: %s", path,
-                           status.c_message());
-  }
-  return base::OkStatus();
+  auto res = ctx->engine->Execute(SqlSource::FromMetricFile(subbed_sql, path));
+  return res.status();
 }
 
 base::Status UnwrapMetricProto::Run(Context*,
@@ -708,7 +704,7 @@
   return base::OkStatus();
 }
 
-base::Status ComputeMetrics(TraceProcessor* tp,
+base::Status ComputeMetrics(PerfettoSqlEngine* engine,
                             const std::vector<std::string> metrics_to_compute,
                             const std::vector<SqlMetricFile>& sql_metrics,
                             const DescriptorPool& pool,
@@ -726,9 +722,9 @@
       return base::ErrStatus("Unknown metric %s", name.c_str());
 
     const auto& sql_metric = *metric_it;
-    auto prep_it = tp->ExecuteQuery(sql_metric.sql);
-    prep_it.Next();
-    RETURN_IF_ERROR(prep_it.Status());
+    auto prep_it =
+        engine->Execute(SqlSource::FromMetric(sql_metric.sql, metric_it->path));
+    RETURN_IF_ERROR(prep_it.status());
 
     auto output_query =
         "SELECT * FROM " + sql_metric.output_table_name.value() + ";";
@@ -736,37 +732,37 @@
         metatrace::Category::QUERY, "COMPUTE_METRIC_QUERY",
         [&](metatrace::Record* r) { r->AddArg("SQL", output_query); });
 
-    auto it = tp->ExecuteQuery(output_query.c_str());
-    auto has_next = it.Next();
-    RETURN_IF_ERROR(it.Status());
+    auto it = engine->ExecuteUntilLastStatement(
+        SqlSource::FromExecuteQuery(output_query));
+    RETURN_IF_ERROR(it.status());
 
     // Allow the query to return no rows. This has the same semantic as an
     // empty proto being returned.
     const auto& field_name = sql_metric.proto_field_name.value();
-    if (!has_next) {
+    if (it->stmt.IsDone()) {
       metric_builder.AppendBytes(field_name, nullptr, 0);
       continue;
     }
 
-    if (it.ColumnCount() != 1) {
+    if (it->stats.column_count != 1) {
       return base::ErrStatus("Output table %s should have exactly one column",
                              sql_metric.output_table_name.value().c_str());
     }
 
-    SqlValue col = it.Get(0);
+    SqlValue col = sqlite_utils::SqliteValueToSqlValue(
+        sqlite3_column_value(it->stmt.sqlite_stmt(), 0));
     if (col.type != SqlValue::kBytes) {
       return base::ErrStatus("Output table %s column has invalid type",
                              sql_metric.output_table_name.value().c_str());
     }
     RETURN_IF_ERROR(metric_builder.AppendSqlValue(field_name, col));
 
-    has_next = it.Next();
+    bool has_next = it->stmt.Step();
     if (has_next) {
       return base::ErrStatus("Output table %s should have at most one row",
                              sql_metric.output_table_name.value().c_str());
     }
-
-    RETURN_IF_ERROR(it.Status());
+    RETURN_IF_ERROR(it->stmt.status());
   }
   *metrics_proto = metric_builder.SerializeRaw();
   return base::OkStatus();
diff --git a/src/trace_processor/metrics/metrics.h b/src/trace_processor/metrics/metrics.h
index 2691e91..889a4d9 100644
--- a/src/trace_processor/metrics/metrics.h
+++ b/src/trace_processor/metrics/metrics.h
@@ -28,6 +28,7 @@
 #include "perfetto/protozero/scattered_heap_buffer.h"
 #include "perfetto/trace_processor/trace_processor.h"
 #include "src/trace_processor/prelude/functions/sql_function.h"
+#include "src/trace_processor/sqlite/perfetto_sql_engine.h"
 #include "src/trace_processor/util/descriptors.h"
 
 #include "protos/perfetto/trace_processor/metrics_impl.pbzero.h"
@@ -175,7 +176,7 @@
 // Implements the RUN_METRIC SQL function.
 struct RunMetric : public SqlFunction {
   struct Context {
-    TraceProcessor* tp;
+    PerfettoSqlEngine* engine;
     std::vector<SqlMetricFile>* metrics;
   };
   static constexpr bool kVoidReturn = true;
@@ -199,7 +200,7 @@
 void RepeatedFieldStep(sqlite3_context* ctx, int argc, sqlite3_value** argv);
 void RepeatedFieldFinal(sqlite3_context* ctx);
 
-base::Status ComputeMetrics(TraceProcessor* impl,
+base::Status ComputeMetrics(PerfettoSqlEngine*,
                             const std::vector<std::string> metrics_to_compute,
                             const std::vector<SqlMetricFile>& metrics,
                             const DescriptorPool& pool,
diff --git a/src/trace_processor/prelude/functions/create_function.cc b/src/trace_processor/prelude/functions/create_function.cc
index f306ad1..a16f949 100644
--- a/src/trace_processor/prelude/functions/create_function.cc
+++ b/src/trace_processor/prelude/functions/create_function.cc
@@ -24,6 +24,7 @@
 #include "src/trace_processor/prelude/functions/create_function_internal.h"
 #include "src/trace_processor/sqlite/perfetto_sql_engine.h"
 #include "src/trace_processor/sqlite/scoped_db.h"
+#include "src/trace_processor/sqlite/sql_source.h"
 #include "src/trace_processor/sqlite/sqlite_engine.h"
 #include "src/trace_processor/sqlite/sqlite_utils.h"
 #include "src/trace_processor/tp_metatrace.h"
@@ -34,20 +35,14 @@
 
 namespace {
 
-base::StatusOr<ScopedStmt> CreateStatement(PerfettoSqlEngine* engine,
-                                           const std::string& sql,
-                                           const std::string& prototype) {
-  ScopedStmt stmt;
-  const char* tail = nullptr;
-  base::Status status = sqlite_utils::PrepareStmt(engine->sqlite_engine()->db(),
-                                                  sql.c_str(), &stmt, &tail);
-  if (!status.ok()) {
-    return base::ErrStatus(
-        "CREATE_FUNCTION[prototype=%s]: SQLite error when preparing "
-        "statement %s",
-        prototype.c_str(), status.message().c_str());
-  }
-  return std::move(stmt);
+base::StatusOr<SqliteEngine::PreparedStatement> CreateStatement(
+    PerfettoSqlEngine* engine,
+    const std::string& sql,
+    const std::string& prototype) {
+  auto res = engine->sqlite_engine()->PrepareStatement(
+      SqlSource::FromFunction(sql.c_str(), prototype));
+  RETURN_IF_ERROR(res.status());
+  return std::move(res.value());
 }
 
 base::Status CheckNoMoreRows(sqlite3_stmt* stmt,
@@ -454,7 +449,7 @@
   // Prepare a statement and push it into the stack of allocated statements
   // for this function.
   base::Status PrepareStatement() {
-    base::StatusOr<ScopedStmt> stmt =
+    base::StatusOr<SqliteEngine::PreparedStatement> stmt =
         CreateStatement(engine_, sql_, prototype_str_);
     RETURN_IF_ERROR(stmt.status());
     is_valid_ = true;
@@ -492,7 +487,7 @@
 
   // Returns the statement that is used for the current invocation.
   sqlite3_stmt* CurrentStatement() {
-    return stmts_[current_recursion_level_ - 1].get();
+    return stmts_[current_recursion_level_ - 1].sqlite_stmt();
   }
 
   // This function is called each time the function returns and resets the
@@ -585,7 +580,7 @@
   // the stack requires a dedicated statement, we maintain a stack of prepared
   // statements and use the top one for each new call (allocating a new one if
   // needed).
-  std::vector<ScopedStmt> stmts_;
+  std::vector<SqliteEngine::PreparedStatement> stmts_;
   // A list of statements to verify to ensure that they don't have more rows
   // in VerifyPostConditions.
   std::vector<sqlite3_stmt*> empty_stmts_to_validate_;
diff --git a/src/trace_processor/prelude/functions/create_view_function.cc b/src/trace_processor/prelude/functions/create_view_function.cc
index 952924f..78f1640 100644
--- a/src/trace_processor/prelude/functions/create_view_function.cc
+++ b/src/trace_processor/prelude/functions/create_view_function.cc
@@ -17,14 +17,15 @@
 #include "src/trace_processor/prelude/functions/create_view_function.h"
 
 #include <numeric>
+#include <optional>
 
 #include "perfetto/base/status.h"
 #include "perfetto/ext/base/string_utils.h"
 #include "perfetto/ext/base/string_view.h"
 #include "perfetto/trace_processor/basic_types.h"
 #include "src/trace_processor/prelude/functions/create_function_internal.h"
+#include "src/trace_processor/sqlite/perfetto_sql_engine.h"
 #include "src/trace_processor/sqlite/scoped_db.h"
-#include "src/trace_processor/sqlite/sqlite_engine.h"
 #include "src/trace_processor/sqlite/sqlite_table.h"
 #include "src/trace_processor/sqlite/sqlite_utils.h"
 #include "src/trace_processor/tp_metatrace.h"
@@ -36,7 +37,7 @@
 namespace {
 
 class CreatedViewFunction final
-    : public TypedSqliteTable<CreatedViewFunction, void*> {
+    : public TypedSqliteTable<CreatedViewFunction, PerfettoSqlEngine*> {
  public:
   class Cursor final : public SqliteTable::BaseCursor {
    public:
@@ -51,14 +52,13 @@
     base::Status Column(sqlite3_context* context, int N);
 
    private:
-    ScopedStmt scoped_stmt_;
-    sqlite3_stmt* stmt_ = nullptr;
+    std::optional<SqliteEngine::PreparedStatement> stmt_;
     CreatedViewFunction* table_ = nullptr;
     bool is_eof_ = false;
     int next_call_count_ = 0;
   };
 
-  CreatedViewFunction(sqlite3*, void*);
+  CreatedViewFunction(sqlite3*, PerfettoSqlEngine*);
   ~CreatedViewFunction() final;
 
   base::Status Init(int argc, const char* const* argv, Schema*) final;
@@ -84,7 +84,7 @@
     return i == (return_values_.size() + prototype_.arguments.size());
   }
 
-  sqlite3* db_ = nullptr;
+  PerfettoSqlEngine* engine_ = nullptr;
 
   Prototype prototype_;
   std::vector<sql_argument::ArgumentDefinition> return_values_;
@@ -93,7 +93,8 @@
   std::string sql_defn_str_;
 };
 
-CreatedViewFunction::CreatedViewFunction(sqlite3* db, void*) : db_(db) {}
+CreatedViewFunction::CreatedViewFunction(sqlite3*, PerfettoSqlEngine* engine)
+    : engine_(engine) {}
 CreatedViewFunction::~CreatedViewFunction() = default;
 
 base::Status CreatedViewFunction::Init(int argc,
@@ -132,30 +133,18 @@
   }
 
   // Verify that the provided SQL prepares to a statement correctly.
-  ScopedStmt stmt;
-  sqlite3_stmt* raw_stmt = nullptr;
-  int ret = sqlite3_prepare_v2(db_, sql_defn_str_.data(),
-                               static_cast<int>(sql_defn_str_.size()),
-                               &raw_stmt, nullptr);
-  stmt.reset(raw_stmt);
-  if (ret != SQLITE_OK) {
-    return base::ErrStatus(
-        "%s: Failed to prepare SQL statement for function. "
-        "Check the SQL defintion this function for syntax errors.\n%s",
-        prototype_.function_name.c_str(),
-        sqlite_utils::FormatErrorMessage(
-            raw_stmt, base::StringView(sql_defn_str_), db_, ret)
-            .c_message());
-  }
+  auto res = engine_->sqlite_engine()->PrepareStatement(
+      SqlSource::FromFunction(sql_defn_str_, prototype_str_));
+  RETURN_IF_ERROR(res.status());
 
   // Verify that every argument name in the function appears in the
   // argument list.
   //
   // We intentionally loop from 1 to |used_param_count| because SQL
   // parameters are 1-indexed *not* 0-indexed.
-  int used_param_count = sqlite3_bind_parameter_count(stmt.get());
+  int used_param_count = sqlite3_bind_parameter_count(res->sqlite_stmt());
   for (int i = 1; i <= used_param_count; ++i) {
-    const char* name = sqlite3_bind_parameter_name(stmt.get(), i);
+    const char* name = sqlite3_bind_parameter_name(res->sqlite_stmt(), i);
 
     if (!name) {
       return base::ErrStatus(
@@ -187,7 +176,8 @@
 
   // Verify that the prepared statement column count matches the return
   // count.
-  uint32_t col_count = static_cast<uint32_t>(sqlite3_column_count(stmt.get()));
+  uint32_t col_count =
+      static_cast<uint32_t>(sqlite3_column_count(res->sqlite_stmt()));
   if (col_count != return_values_.size()) {
     return base::ErrStatus(
         "%s: number of return values %u does not match SQL statement column "
@@ -197,7 +187,8 @@
 
   // Verify that the return names matches the prepared statment column names.
   for (uint32_t i = 0; i < col_count; ++i) {
-    const char* name = sqlite3_column_name(stmt.get(), static_cast<int>(i));
+    const char* name =
+        sqlite3_column_name(res->sqlite_stmt(), static_cast<int>(i));
     if (name != return_values_[i].name()) {
       return base::ErrStatus(
           "%s: column %s at index %u does not match return value name %s.",
@@ -330,11 +321,10 @@
   // creating it very time.
   // TODO(lalitm): measure and implement whether it would be a good idea to
   // forward constraints here when we build the nested query.
-  int ret = sqlite3_prepare_v2(table_->db_, table_->sql_defn_str_.data(),
-                               static_cast<int>(table_->sql_defn_str_.size()),
-                               &stmt_, nullptr);
-  scoped_stmt_.reset(stmt_);
-  PERFETTO_CHECK(ret == SQLITE_OK);
+  auto res = table_->engine_->sqlite_engine()->PrepareStatement(
+      SqlSource::FromFunction(table_->sql_defn_str_, table_->prototype_str_));
+  RETURN_IF_ERROR(res->status());
+  stmt_ = std::move(res.value());
 
   // Bind all the arguments to the appropriate places in the function.
   for (size_t i = 0; i < qc.constraints().size(); ++i) {
@@ -351,8 +341,8 @@
     PERFETTO_DCHECK(index < table_->prototype_.arguments.size());
 
     const auto& arg = table_->prototype_.arguments[index];
-    auto status = MaybeBindArgument(stmt_, table_->prototype_.function_name,
-                                    arg, argv[i]);
+    auto status = MaybeBindArgument(
+        stmt_->sqlite_stmt(), table_->prototype_.function_name, arg, argv[i]);
     RETURN_IF_ERROR(status);
   }
 
@@ -363,17 +353,9 @@
 }
 
 base::Status CreatedViewFunction::Cursor::Next() {
-  int ret = sqlite3_step(stmt_);
-  is_eof_ = ret == SQLITE_DONE;
+  is_eof_ = !stmt_->Step();
   next_call_count_++;
-  if (ret != SQLITE_ROW && ret != SQLITE_DONE) {
-    return base::ErrStatus(
-        "%s: SQLite error while stepping statement: %s",
-        table_->prototype_.function_name.c_str(),
-        sqlite_utils::FormatErrorMessage(stmt_, std::nullopt, table_->db_, ret)
-            .c_message());
-  }
-  return base::OkStatus();
+  return stmt_->status();
 }
 
 bool CreatedViewFunction::Cursor::Eof() {
@@ -383,7 +365,7 @@
 base::Status CreatedViewFunction::Cursor::Column(sqlite3_context* ctx, int i) {
   size_t idx = static_cast<size_t>(i);
   if (table_->IsReturnValueColumn(idx)) {
-    sqlite3_result_value(ctx, sqlite3_column_value(stmt_, i));
+    sqlite3_result_value(ctx, sqlite3_column_value(stmt_->sqlite_stmt(), i));
   } else if (table_->IsArgumentColumn(idx)) {
     // TODO(lalitm): it may be more appropriate to keep a note of the arguments
     // which we passed in and return them here. Not doing this to because it
@@ -455,36 +437,21 @@
   )""";
   std::string function_name_str = function_name.ToStdString();
 
-  ScopedSqliteString errmsg;
-  char* errmsg_raw = nullptr;
-  int ret;
-
   NullTermStringView sql_defn(sql_defn_str);
-  if (sql_defn.size() < 512) {
-    base::StackString<1024> sql(kSqlTemplate, function_name_str.c_str(),
-                                function_name_str.c_str(), prototype_str,
-                                return_prototype_str, sql_defn_str);
-    ret = sqlite3_exec(ctx->db, sql.c_str(), nullptr, nullptr, &errmsg_raw);
-  } else {
-    std::vector<char> formatted_sql(sql_defn.size() + 1024);
-    base::SprintfTrunc(formatted_sql.data(), formatted_sql.size(), kSqlTemplate,
-                       function_name_str.c_str(), function_name_str.c_str(),
-                       prototype_str, return_prototype_str, sql_defn_str);
-    ret = sqlite3_exec(ctx->db, formatted_sql.data(), nullptr, nullptr,
-                       &errmsg_raw);
-  }
-
-  errmsg.reset(errmsg_raw);
-  if (ret != SQLITE_OK)
-    return base::ErrStatus("%s", errmsg.get());
+  std::string formatted_sql(sql_defn.size() + 1024, '\0');
+  base::SprintfTrunc(formatted_sql.data(), formatted_sql.size(), kSqlTemplate,
+                     function_name_str.c_str(), function_name_str.c_str(),
+                     prototype_str, return_prototype_str, sql_defn_str);
 
   // CREATE_VIEW_FUNCTION doesn't have a return value so just don't sent |out|.
-  return base::OkStatus();
+  auto res = ctx->Execute(
+      SqlSource::FromFunction(std::move(formatted_sql), prototype_str));
+  return res.status();
 }
 
-void RegisterCreateViewFunctionModule(SqliteEngine* engine) {
-  engine->RegisterVirtualTableModule<CreatedViewFunction>(
-      "internal_view_function_impl", nullptr,
+void RegisterCreateViewFunctionModule(PerfettoSqlEngine* engine) {
+  engine->sqlite_engine()->RegisterVirtualTableModule<CreatedViewFunction>(
+      "internal_view_function_impl", engine,
       SqliteTable::TableType::kExplicitCreate, false);
 }
 
diff --git a/src/trace_processor/prelude/functions/create_view_function.h b/src/trace_processor/prelude/functions/create_view_function.h
index 509ea5e..50ff80f 100644
--- a/src/trace_processor/prelude/functions/create_view_function.h
+++ b/src/trace_processor/prelude/functions/create_view_function.h
@@ -25,15 +25,13 @@
 namespace perfetto {
 namespace trace_processor {
 
-class SqliteEngine;
+class PerfettoSqlEngine;
 
 // Implementation of CREATE_VIEW_FUNCTION SQL function.
 // See https://perfetto.dev/docs/analysis/metrics#metric-helper-functions for
 // usage of this function.
 struct CreateViewFunction : public SqlFunction {
-  struct Context {
-    sqlite3* db;
-  };
+  using Context = PerfettoSqlEngine;
 
   static constexpr bool kVoidReturn = true;
 
@@ -44,7 +42,7 @@
                           Destructors&);
 };
 
-void RegisterCreateViewFunctionModule(SqliteEngine*);
+void RegisterCreateViewFunctionModule(PerfettoSqlEngine*);
 
 }  // namespace trace_processor
 }  // namespace perfetto
diff --git a/src/trace_processor/prelude/functions/import.cc b/src/trace_processor/prelude/functions/import.cc
index 7657969..fd47a9c 100644
--- a/src/trace_processor/prelude/functions/import.cc
+++ b/src/trace_processor/prelude/functions/import.cc
@@ -24,6 +24,7 @@
 #include "perfetto/trace_processor/basic_types.h"
 #include "src/trace_processor/prelude/functions/create_function_internal.h"
 #include "src/trace_processor/sqlite/scoped_db.h"
+#include "src/trace_processor/sqlite/sql_source.h"
 #include "src/trace_processor/sqlite/sqlite_table.h"
 #include "src/trace_processor/sqlite/sqlite_utils.h"
 #include "src/trace_processor/tp_metatrace.h"
@@ -75,15 +76,11 @@
     return base::OkStatus();
   }
 
-  auto import_iter = ctx->tp->ExecuteQuery(module_file->sql);
-  if (import_iter.StatementWithOutputCount() > 0)
+  auto it = ctx->engine->Execute(
+      SqlSource::FromModuleImport(module_file->sql, import_key));
+  RETURN_IF_ERROR(it.status());
+  if (it->statement_count_with_output > 0)
     return base::ErrStatus("IMPORT: Imported file returning values.");
-  {
-    auto status = import_iter.Status();
-    if (!status.ok())
-      return base::ErrStatus("SQLite error on IMPORT: %s", status.c_message());
-  }
-
   module_file->imported = true;
   return base::OkStatus();
 }
diff --git a/src/trace_processor/prelude/functions/import.h b/src/trace_processor/prelude/functions/import.h
index da40f5f..bd53c7d 100644
--- a/src/trace_processor/prelude/functions/import.h
+++ b/src/trace_processor/prelude/functions/import.h
@@ -24,6 +24,7 @@
 #include "perfetto/ext/base/flat_hash_map.h"
 #include "perfetto/trace_processor/trace_processor.h"
 #include "src/trace_processor/prelude/functions/sql_function.h"
+#include "src/trace_processor/sqlite/perfetto_sql_engine.h"
 #include "src/trace_processor/util/sql_modules.h"
 
 namespace perfetto {
@@ -31,8 +32,7 @@
 
 struct Import : public SqlFunction {
   struct Context {
-    sqlite3* db;
-    TraceProcessor* tp;
+    PerfettoSqlEngine* engine;
     base::FlatHashMap<std::string, sql_modules::RegisteredModule>* modules;
   };
 
diff --git a/src/trace_processor/prelude/operators/span_join_operator.cc b/src/trace_processor/prelude/operators/span_join_operator.cc
index 2ee2839..153166c 100644
--- a/src/trace_processor/prelude/operators/span_join_operator.cc
+++ b/src/trace_processor/prelude/operators/span_join_operator.cc
@@ -28,6 +28,7 @@
 #include "perfetto/ext/base/string_splitter.h"
 #include "perfetto/ext/base/string_utils.h"
 #include "perfetto/ext/base/string_view.h"
+#include "src/trace_processor/sqlite/perfetto_sql_engine.h"
 #include "src/trace_processor/sqlite/sqlite_utils.h"
 #include "src/trace_processor/tp_metatrace.h"
 #include "src/trace_processor/util/status_macros.h"
@@ -106,8 +107,9 @@
 
 }  // namespace
 
-SpanJoinOperatorTable::SpanJoinOperatorTable(sqlite3* db, const void*)
-    : db_(db) {}
+SpanJoinOperatorTable::SpanJoinOperatorTable(sqlite3*,
+                                             PerfettoSqlEngine* engine)
+    : engine_(engine) {}
 SpanJoinOperatorTable::~SpanJoinOperatorTable() = default;
 
 util::Status SpanJoinOperatorTable::Init(int argc,
@@ -229,7 +231,8 @@
 }
 
 std::unique_ptr<SqliteTable::BaseCursor> SpanJoinOperatorTable::CreateCursor() {
-  return std::unique_ptr<SpanJoinOperatorTable::Cursor>(new Cursor(this, db_));
+  return std::unique_ptr<SpanJoinOperatorTable::Cursor>(
+      new Cursor(this, engine_));
 }
 
 int SpanJoinOperatorTable::BestIndex(const QueryConstraints& qc,
@@ -328,10 +331,8 @@
   }
 
   std::vector<SqliteTable::Column> cols;
-  auto status = sqlite_utils::GetColumnsForTable(db_, desc.name, cols);
-  if (!status.ok()) {
-    return status;
-  }
+  RETURN_IF_ERROR(sqlite_utils::GetColumnsForTable(
+      engine_->sqlite_engine()->db(), desc.name, cols));
 
   uint32_t required_columns_found = 0;
   uint32_t ts_idx = std::numeric_limits<uint32_t>::max();
@@ -392,10 +393,11 @@
   return defn.columns()[locator.col_index].name().c_str();
 }
 
-SpanJoinOperatorTable::Cursor::Cursor(SpanJoinOperatorTable* table, sqlite3* db)
+SpanJoinOperatorTable::Cursor::Cursor(SpanJoinOperatorTable* table,
+                                      PerfettoSqlEngine* engine)
     : SqliteTable::BaseCursor(table),
-      t1_(table, &table->t1_defn_, db),
-      t2_(table, &table->t2_defn_, db),
+      t1_(table, &table->t1_defn_, engine),
+      t2_(table, &table->t2_defn_, engine),
       table_(table) {}
 SpanJoinOperatorTable::Cursor::~Cursor() = default;
 
@@ -592,8 +594,8 @@
 
 SpanJoinOperatorTable::Query::Query(SpanJoinOperatorTable* table,
                                     const TableDefinition* definition,
-                                    sqlite3* db)
-    : defn_(definition), db_(db), table_(table) {
+                                    PerfettoSqlEngine* engine)
+    : defn_(definition), engine_(engine), table_(table) {
   PERFETTO_DCHECK(!defn_->IsPartitioned() ||
                   defn_->partition_idx() < defn_->columns().size());
 }
@@ -604,7 +606,7 @@
     const QueryConstraints& qc,
     sqlite3_value** argv,
     InitialEofBehavior eof_behavior) {
-  *this = Query(table_, definition(), db_);
+  *this = Query(table_, definition(), engine_);
   sql_query_ = CreateSqlQuery(
       table_->ComputeSqlConstraintsForDefinition(*defn_, qc, argv));
   util::Status status = Rewind();
@@ -729,18 +731,11 @@
 }
 
 util::Status SpanJoinOperatorTable::Query::Rewind() {
-  sqlite3_stmt* stmt = nullptr;
-  int res =
-      sqlite3_prepare_v2(db_, sql_query_.c_str(),
-                         static_cast<int>(sql_query_.size()), &stmt, nullptr);
-  stmt_.reset(stmt);
-
-  cursor_eof_ = res != SQLITE_OK;
-  if (res != SQLITE_OK)
-    return util::ErrStatus(
-        "%s", sqlite_utils::FormatErrorMessage(
-                  stmt_.get(), base::StringView(sql_query_), db_, res)
-                  .c_message());
+  auto res = engine_->sqlite_engine()->PrepareStatement(
+      SqlSource::FromSpanJoin(sql_query_, table_->name()));
+  cursor_eof_ = false;
+  RETURN_IF_ERROR(res.status());
+  stmt_ = std::move(res.value());
 
   RETURN_IF_ERROR(CursorNext());
 
@@ -765,27 +760,23 @@
 }
 
 util::Status SpanJoinOperatorTable::Query::CursorNext() {
-  auto* stmt = stmt_.get();
-  int res;
   if (defn_->IsPartitioned()) {
     auto partition_idx = static_cast<int>(defn_->partition_idx());
     // Fastforward through any rows with null partition keys.
     int row_type;
     do {
-      res = sqlite3_step(stmt);
-      row_type = sqlite3_column_type(stmt, partition_idx);
-    } while (res == SQLITE_ROW && row_type == SQLITE_NULL);
+      cursor_eof_ = !stmt_->Step();
+      RETURN_IF_ERROR(stmt_->status());
+      row_type = sqlite3_column_type(stmt_->sqlite_stmt(), partition_idx);
+    } while (!cursor_eof_ && row_type == SQLITE_NULL);
 
-    if (res == SQLITE_ROW && row_type != SQLITE_INTEGER) {
+    if (!cursor_eof_ && row_type != SQLITE_INTEGER) {
       return util::ErrStatus("SPAN_JOIN: partition is not an int");
     }
   } else {
-    res = sqlite3_step(stmt);
+    cursor_eof_ = !stmt_->Step();
   }
-  cursor_eof_ = res != SQLITE_ROW;
-  return res == SQLITE_ROW || res == SQLITE_DONE
-             ? util::OkStatus()
-             : util::ErrStatus("SPAN_JOIN: %s", sqlite3_errmsg(db_));
+  return base::OkStatus();
 }
 
 std::string SpanJoinOperatorTable::Query::CreateSqlQuery(
@@ -816,7 +807,7 @@
     return;
   }
 
-  sqlite3_stmt* stmt = stmt_.get();
+  sqlite3_stmt* stmt = stmt_->sqlite_stmt();
   int idx = static_cast<int>(index);
   switch (sqlite3_column_type(stmt, idx)) {
     case SQLITE_INTEGER:
diff --git a/src/trace_processor/prelude/operators/span_join_operator.h b/src/trace_processor/prelude/operators/span_join_operator.h
index 3f7a006..8b442ed 100644
--- a/src/trace_processor/prelude/operators/span_join_operator.h
+++ b/src/trace_processor/prelude/operators/span_join_operator.h
@@ -24,6 +24,7 @@
 #include <limits>
 #include <map>
 #include <memory>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -32,11 +33,14 @@
 #include "perfetto/trace_processor/basic_types.h"
 #include "perfetto/trace_processor/status.h"
 #include "src/trace_processor/sqlite/scoped_db.h"
+#include "src/trace_processor/sqlite/sqlite_engine.h"
 #include "src/trace_processor/sqlite/sqlite_table.h"
 
 namespace perfetto {
 namespace trace_processor {
 
+class PerfettoSqlEngine;
+
 // Implements the SPAN JOIN operation between two tables on a particular column.
 //
 // Span:
@@ -70,7 +74,7 @@
 // All other columns apart from timestamp (ts), duration (dur) and the join key
 // are passed through unchanged.
 class SpanJoinOperatorTable final
-    : public TypedSqliteTable<SpanJoinOperatorTable, const void*> {
+    : public TypedSqliteTable<SpanJoinOperatorTable, PerfettoSqlEngine*> {
  public:
   // Enum indicating whether the queries on the two inner tables should
   // emit shadows.
@@ -159,7 +163,9 @@
       kEof,
     };
 
-    Query(SpanJoinOperatorTable*, const TableDefinition*, sqlite3* db);
+    Query(SpanJoinOperatorTable*,
+          const TableDefinition*,
+          PerfettoSqlEngine* engine);
     virtual ~Query();
 
     Query(Query&&) noexcept = default;
@@ -278,20 +284,20 @@
     int64_t CursorTs() const {
       PERFETTO_DCHECK(!cursor_eof_);
       auto ts_idx = static_cast<int>(defn_->ts_idx());
-      return sqlite3_column_int64(stmt_.get(), ts_idx);
+      return sqlite3_column_int64(stmt_->sqlite_stmt(), ts_idx);
     }
 
     int64_t CursorDur() const {
       PERFETTO_DCHECK(!cursor_eof_);
       auto dur_idx = static_cast<int>(defn_->dur_idx());
-      return sqlite3_column_int64(stmt_.get(), dur_idx);
+      return sqlite3_column_int64(stmt_->sqlite_stmt(), dur_idx);
     }
 
     int64_t CursorPartition() const {
       PERFETTO_DCHECK(!cursor_eof_);
       PERFETTO_DCHECK(defn_->IsPartitioned());
       auto partition_idx = static_cast<int>(defn_->partition_idx());
-      return sqlite3_column_int64(stmt_.get(), partition_idx);
+      return sqlite3_column_int64(stmt_->sqlite_stmt(), partition_idx);
     }
 
     State state_ = State::kMissingPartitionShadow;
@@ -309,17 +315,17 @@
     int64_t missing_partition_end_ = 0;
 
     std::string sql_query_;
-    ScopedStmt stmt_;
+    std::optional<SqliteEngine::PreparedStatement> stmt_;
 
     const TableDefinition* defn_ = nullptr;
-    sqlite3* db_ = nullptr;
+    PerfettoSqlEngine* engine_ = nullptr;
     SpanJoinOperatorTable* table_ = nullptr;
   };
 
   // Base class for a cursor on the span table.
   class Cursor final : public SqliteTable::BaseCursor {
    public:
-    Cursor(SpanJoinOperatorTable*, sqlite3* db);
+    Cursor(SpanJoinOperatorTable*, PerfettoSqlEngine*);
     ~Cursor() final;
 
     base::Status Filter(const QueryConstraints& qc,
@@ -351,7 +357,7 @@
     SpanJoinOperatorTable* table_;
   };
 
-  SpanJoinOperatorTable(sqlite3*, const void*);
+  SpanJoinOperatorTable(sqlite3*, PerfettoSqlEngine*);
   ~SpanJoinOperatorTable() final;
 
   // Table implementation.
@@ -431,7 +437,7 @@
   PartitioningType partitioning_;
   base::FlatHashMap<size_t, ColumnLocator> global_index_to_column_locator_;
 
-  sqlite3* const db_;
+  PerfettoSqlEngine* engine_ = nullptr;
 };
 
 }  // namespace trace_processor
diff --git a/src/trace_processor/prelude/operators/span_join_operator_unittest.cc b/src/trace_processor/prelude/operators/span_join_operator_unittest.cc
index 661a7f1..6c76f34 100644
--- a/src/trace_processor/prelude/operators/span_join_operator_unittest.cc
+++ b/src/trace_processor/prelude/operators/span_join_operator_unittest.cc
@@ -16,6 +16,7 @@
 
 #include "src/trace_processor/prelude/operators/span_join_operator.h"
 
+#include "src/trace_processor/sqlite/perfetto_sql_engine.h"
 #include "src/trace_processor/sqlite/sqlite_engine.h"
 #include "test/gtest_and_gmock.h"
 
@@ -26,19 +27,19 @@
 class SpanJoinOperatorTableTest : public ::testing::Test {
  public:
   SpanJoinOperatorTableTest() {
-    engine_.RegisterVirtualTableModule<SpanJoinOperatorTable>(
-        "span_join", nullptr, SqliteTable::TableType::kExplicitCreate, false);
-    engine_.RegisterVirtualTableModule<SpanJoinOperatorTable>(
-        "span_left_join", nullptr, SqliteTable::TableType::kExplicitCreate,
+    engine_.sqlite_engine()->RegisterVirtualTableModule<SpanJoinOperatorTable>(
+        "span_join", &engine_, SqliteTable::TableType::kExplicitCreate, false);
+    engine_.sqlite_engine()->RegisterVirtualTableModule<SpanJoinOperatorTable>(
+        "span_left_join", &engine_, SqliteTable::TableType::kExplicitCreate,
         false);
   }
 
   void PrepareValidStatement(const std::string& sql) {
     int size = static_cast<int>(sql.size());
     sqlite3_stmt* stmt;
-    ASSERT_EQ(
-        sqlite3_prepare_v2(engine_.db(), sql.c_str(), size, &stmt, nullptr),
-        SQLITE_OK);
+    ASSERT_EQ(sqlite3_prepare_v2(engine_.sqlite_engine()->db(), sql.c_str(),
+                                 size, &stmt, nullptr),
+              SQLITE_OK);
     stmt_.reset(stmt);
   }
 
@@ -56,7 +57,7 @@
   }
 
  protected:
-  SqliteEngine engine_;
+  PerfettoSqlEngine engine_;
   ScopedStmt stmt_;
 };
 
diff --git a/src/trace_processor/rpc/query_result_serializer_unittest.cc b/src/trace_processor/rpc/query_result_serializer_unittest.cc
index 6344edc..8038bae 100644
--- a/src/trace_processor/rpc/query_result_serializer_unittest.cc
+++ b/src/trace_processor/rpc/query_result_serializer_unittest.cc
@@ -408,7 +408,9 @@
   TestDeserializer deser;
   deser.SerializeAndDeserialize(&ser);
   EXPECT_EQ(deser.cells.size(), 0u);
-  EXPECT_EQ(deser.error, "Error: incomplete input (errcode: 1)");
+  EXPECT_EQ(deser.error,
+            "Traceback (most recent call last):\n  File \"stdin\" line 1 col "
+            "1\n    insert into incomplete_input\n    ^\nincomplete input");
   EXPECT_TRUE(deser.eof_reached);
 }
 
diff --git a/src/trace_processor/sqlite/BUILD.gn b/src/trace_processor/sqlite/BUILD.gn
index 5ba1283..6904f2c 100644
--- a/src/trace_processor/sqlite/BUILD.gn
+++ b/src/trace_processor/sqlite/BUILD.gn
@@ -26,6 +26,8 @@
     "perfetto_sql_parser.h",
     "query_cache.h",
     "scoped_db.h",
+    "sql_source.cc",
+    "sql_source.h",
     "sql_stats_table.cc",
     "sql_stats_table.h",
     "sqlite_engine.cc",
diff --git a/src/trace_processor/sqlite/perfetto_sql_engine.cc b/src/trace_processor/sqlite/perfetto_sql_engine.cc
index ea06188..c174814 100644
--- a/src/trace_processor/sqlite/perfetto_sql_engine.cc
+++ b/src/trace_processor/sqlite/perfetto_sql_engine.cc
@@ -16,7 +16,18 @@
 
 #include "src/trace_processor/sqlite/perfetto_sql_engine.h"
 
+#include <optional>
+#include <string>
+#include <variant>
+
+#include "perfetto/base/status.h"
+#include "perfetto/ext/base/string_utils.h"
 #include "src/trace_processor/sqlite/db_sqlite_table.h"
+#include "src/trace_processor/sqlite/perfetto_sql_parser.h"
+#include "src/trace_processor/sqlite/scoped_db.h"
+#include "src/trace_processor/sqlite/sql_source.h"
+#include "src/trace_processor/sqlite/sqlite_engine.h"
+#include "src/trace_processor/sqlite/sqlite_utils.h"
 #include "src/trace_processor/tp_metatrace.h"
 #include "src/trace_processor/util/status_macros.h"
 
@@ -24,14 +35,15 @@
 namespace trace_processor {
 namespace {
 
-void IncrementCountForStmt(sqlite3_stmt* stmt,
-                           PerfettoSqlEngine::ExecutionResult* res) {
+void IncrementCountForStmt(const SqliteEngine::PreparedStatement& p_stmt,
+                           PerfettoSqlEngine::ExecutionStats* res) {
   res->statement_count++;
 
   // If the stmt is already done, it clearly didn't have any output.
-  if (sqlite_utils::IsStmtDone(stmt))
+  if (p_stmt.IsDone())
     return;
 
+  sqlite3_stmt* stmt = p_stmt.sqlite_stmt();
   if (sqlite3_column_count(stmt) == 1) {
     sqlite3_value* value = sqlite3_column_value(stmt, 0);
 
@@ -94,100 +106,101 @@
       table_name, std::move(context), SqliteTable::kEponymousOnly, false);
 }
 
-base::StatusOr<PerfettoSqlEngine::ExecutionResult>
-PerfettoSqlEngine::ExecuteUntilLastStatement(const std::string& sql) {
-  ExecutionResult res;
+base::StatusOr<PerfettoSqlEngine::ExecutionStats> PerfettoSqlEngine::Execute(
+    SqlSource sql) {
+  auto res = ExecuteUntilLastStatement(std::move(sql));
+  RETURN_IF_ERROR(res.status());
+  if (res->stmt.IsDone()) {
+    return res->stats;
+  }
+  while (res->stmt.Step()) {
+  }
+  RETURN_IF_ERROR(res->stmt.status());
+  return res->stats;
+}
 
-  // A sql string can contain several statements. Some of them might be comment
-  // only, e.g. "SELECT 1; /* comment */; SELECT 2;". Here we process one
-  // statement on each iteration. SQLite's sqlite_prepare_v2 (wrapped by
-  // PrepareStmt) returns on each iteration a pointer to the unprocessed string.
-  //
-  // Unfortunately we cannot call PrepareStmt and tokenize all statements
-  // upfront because sqlite_prepare_v2 also semantically checks the statement
-  // against the schema. In some cases statements might depend on the execution
-  // of previous ones (e.e. CREATE VIEW x; SELECT FROM x; DELETE VIEW x;).
-  //
-  // Also, unfortunately, we need to PrepareStmt to find out if a statement is a
-  // comment or a real statement.
+base::StatusOr<PerfettoSqlEngine::ExecutionResult>
+PerfettoSqlEngine::ExecuteUntilLastStatement(SqlSource sql_source) {
+  // A SQL string can contain several statements. Some of them might be comment
+  // only, e.g. "SELECT 1; /* comment */; SELECT 2;". Some statements can also
+  // be PerfettoSQL statements which we need to transpile before execution or
+  // execute without delegating to SQLite.
   //
   // The logic here is the following:
-  //  - We invoke PrepareStmt on each statement.
-  //  - If the statement is a comment we simply skip it.
-  //  - If the statement is valid, we step once to make sure side effects take
-  //    effect.
+  //  - We parse the statement as a PerfettoSQL statement.
+  //  - If the statement is actually an SQLite statement, we invoke PrepareStmt.
+  //  - We step once to make sure side effects take effect (e.g. for CREATE
+  //    TABLE statements, tables are created).
   //  - If we encounter a valid statement afterwards, we step internally through
   //    all rows of the previous one. This ensures that any further side effects
   //    take hold *before* we step into the next statement.
-  //  - Once no further non-comment statements are encountered, we return an
-  //    iterator to the last valid statement.
-  for (const char* rem_sql = sql.c_str(); rem_sql && rem_sql[0];) {
-    ScopedStmt cur_stmt;
+  //  - Once no further statements are encountered, we return the prepared
+  //    statement for the last valid statement.
+  std::optional<SqliteEngine::PreparedStatement> res;
+  ExecutionStats stats;
+  PerfettoSqlParser parser(std::move(sql_source));
+  while (parser.Next()) {
+    // If none of the above matched, this must just be an SQL statement directly
+    // executable by SQLite.
+    auto* sql = std::get_if<PerfettoSqlParser::SqliteSql>(&parser.statement());
+    PERFETTO_CHECK(sql);
+
+    // Try to get SQLite to prepare the statement.
+    std::optional<SqliteEngine::PreparedStatement> cur_stmt;
     {
       PERFETTO_TP_TRACE(metatrace::Category::QUERY, "QUERY_PREPARE");
-      const char* tail = nullptr;
-      RETURN_IF_ERROR(
-          sqlite_utils::PrepareStmt(engine_.db(), rem_sql, &cur_stmt, &tail));
-      rem_sql = tail;
+      auto stmt_or = engine_.PrepareStatement(std::move(sql->sql));
+      RETURN_IF_ERROR(stmt_or.status());
+      cur_stmt = std::move(stmt_or.value());
     }
 
     // The only situation where we'd have an ok status but also no prepared
-    // statement is if the statement of SQL we parsed was a pure comment. In
-    // this case, just continue to the next statement.
-    if (!cur_stmt)
-      continue;
+    // statement is if the SQL was a pure comment. However, the PerfettoSQL
+    // parser should filter out such statements so this should never happen.
+    PERFETTO_DCHECK(cur_stmt->sqlite_stmt());
 
     // Before stepping into |cur_stmt|, we need to finish iterating through
     // the previous statement so we don't have two clashing statements (e.g.
     // SELECT * FROM v and DROP VIEW v) partially stepped into.
-    if (res.stmt) {
+    if (res && !res->IsDone()) {
       PERFETTO_TP_TRACE(metatrace::Category::QUERY, "STMT_STEP_UNTIL_DONE",
                         [&res](metatrace::Record* record) {
-                          auto expanded_sql =
-                              sqlite_utils::ExpandedSqlForStmt(res.stmt.get());
-                          record->AddArg("SQL", expanded_sql.get());
+                          record->AddArg("SQL", res->expanded_sql());
                         });
-      RETURN_IF_ERROR(sqlite_utils::StepStmtUntilDone(res.stmt.get()));
-      res.stmt.reset();
+      while (res->Step()) {
+      }
+      RETURN_IF_ERROR(res->status());
     }
 
-    PERFETTO_DLOG("Executing statement: %s", sqlite3_sql(*cur_stmt));
+    // Propogate the current statement to the next iteration.
+    res = std::move(cur_stmt);
 
+    // Step the newly prepared statement once. This is considered to be
+    // "executing" the statement.
     {
       PERFETTO_TP_TRACE(metatrace::Category::TOPLEVEL, "STMT_FIRST_STEP",
-                        [&cur_stmt](metatrace::Record* record) {
-                          auto expanded_sql =
-                              sqlite_utils::ExpandedSqlForStmt(*cur_stmt);
-                          record->AddArg("SQL", expanded_sql.get());
+                        [&res](metatrace::Record* record) {
+                          record->AddArg("SQL", res->expanded_sql());
                         });
-
-      // Now step once into |cur_stmt| so that when we prepare the next statment
-      // we will have executed any dependent bytecode in this one.
-      int err = sqlite3_step(*cur_stmt);
-      if (err != SQLITE_ROW && err != SQLITE_DONE) {
-        return base::ErrStatus(
-            "%s", sqlite_utils::FormatErrorMessage(
-                      cur_stmt.get(), base::StringView(sql), engine_.db(), err)
-                      .c_message());
-      }
+      PERFETTO_DLOG("Executing statement: %s", res->sql());
+      res->Step();
+      RETURN_IF_ERROR(res->status());
     }
 
     // Increment the neecessary counts for the statement.
-    IncrementCountForStmt(cur_stmt.get(), &res);
-
-    // Propogate the current statement to the next iteration.
-    res.stmt = std::move(cur_stmt);
+    IncrementCountForStmt(*res, &stats);
   }
+  RETURN_IF_ERROR(parser.status());
 
-  // If we didn't manage to prepare a single statment, that means everything
+  // If we didn't manage to prepare a single statement, that means everything
   // in the SQL was treated as a comment.
-  if (!res.stmt)
+  if (!res)
     return base::ErrStatus("No valid SQL to run");
 
-  // Update the output statment and column count.
-  res.column_count =
-      static_cast<uint32_t>(sqlite3_column_count(res.stmt.get()));
-  return std::move(res);
+  // Update the output statement and column count.
+  stats.column_count =
+      static_cast<uint32_t>(sqlite3_column_count(res->sqlite_stmt()));
+  return ExecutionResult{std::move(*res), stats};
 }
 
 }  // namespace trace_processor
diff --git a/src/trace_processor/sqlite/perfetto_sql_engine.h b/src/trace_processor/sqlite/perfetto_sql_engine.h
index f06b159..185c7fe 100644
--- a/src/trace_processor/sqlite/perfetto_sql_engine.h
+++ b/src/trace_processor/sqlite/perfetto_sql_engine.h
@@ -19,6 +19,7 @@
 
 #include "perfetto/ext/base/status_or.h"
 #include "src/trace_processor/sqlite/scoped_db.h"
+#include "src/trace_processor/sqlite/sql_source.h"
 #include "src/trace_processor/sqlite/sqlite_engine.h"
 #include "src/trace_processor/sqlite/sqlite_utils.h"
 
@@ -30,23 +31,34 @@
 // by and executed against SQLite.
 class PerfettoSqlEngine {
  public:
-  struct ExecutionResult {
-    ScopedStmt stmt;
+  struct ExecutionStats {
     uint32_t column_count = 0;
     uint32_t statement_count = 0;
     uint32_t statement_count_with_output = 0;
   };
+  struct ExecutionResult {
+    SqliteEngine::PreparedStatement stmt;
+    ExecutionStats stats;
+  };
 
   PerfettoSqlEngine();
 
-  // Executes all the statements in |sql| until the last one and returns a
-  // |ExecutionResult| object containing a |ScopedStmt| for the final statement
-  // and metadata about all statements executed.
+  // Executes all the statements in |sql| and returns a |ExecutionResult|
+  // object. The metadata will reference all the statements executed and the
+  // |ScopedStmt| be empty.
   //
   // Returns an error if the execution of any statement failed or if there was
   // no valid SQL to run.
-  base::StatusOr<ExecutionResult> ExecuteUntilLastStatement(
-      const std::string& sql);
+  base::StatusOr<ExecutionStats> Execute(SqlSource sql);
+
+  // Executes all the statements in |sql| fully until the final statement and
+  // returns a |ExecutionResult| object containing a |ScopedStmt| for the final
+  // statement (which has been stepped once) and metadata about all statements
+  // executed.
+  //
+  // Returns an error if the execution of any statement failed or if there was
+  // no valid SQL to run.
+  base::StatusOr<ExecutionResult> ExecuteUntilLastStatement(SqlSource sql);
 
   // Registers a trace processor C++ function to be runnable from SQL.
   //
diff --git a/src/trace_processor/sqlite/perfetto_sql_parser.cc b/src/trace_processor/sqlite/perfetto_sql_parser.cc
index fb14368..b2e0ea9 100644
--- a/src/trace_processor/sqlite/perfetto_sql_parser.cc
+++ b/src/trace_processor/sqlite/perfetto_sql_parser.cc
@@ -16,6 +16,7 @@
 
 #include "src/trace_processor/sqlite/perfetto_sql_parser.h"
 #include "perfetto/base/logging.h"
+#include "src/trace_processor/sqlite/sql_source.h"
 #include "src/trace_processor/sqlite/sqlite_tokenizer.h"
 
 namespace perfetto {
@@ -31,8 +32,8 @@
 
 }  // namespace
 
-PerfettoSqlParser::PerfettoSqlParser(const char* sql)
-    : tokenizer_(sql), start_(sql) {}
+PerfettoSqlParser::PerfettoSqlParser(SqlSource sql)
+    : sql_(std::move(sql)), tokenizer_(sql_.sql().c_str()) {}
 
 bool PerfettoSqlParser::Next() {
   PERFETTO_DCHECK(status_.ok());
@@ -47,15 +48,14 @@
 
     if (TokenIsTerminal(token)) {
       // If we have a non-space character we've seen, just return all the stuff
-      // we've seen between that and the current token.
+      // after that point.
       if (non_space_ptr) {
         uint32_t offset_of_non_space =
-            static_cast<uint32_t>(non_space_ptr - start_);
+            static_cast<uint32_t>(non_space_ptr - sql_.sql().c_str());
         uint32_t chars_since_non_space =
             static_cast<uint32_t>(tokenizer_.ptr() - non_space_ptr);
-        statement_ = Statement(
-            SqliteSql{std::string_view(non_space_ptr, chars_since_non_space),
-                      offset_of_non_space});
+        statement_ =
+            SqliteSql{sql_.Substr(offset_of_non_space, chars_since_non_space)};
         return true;
       }
       // This means we've seen a semi-colon without any non-space content. Just
diff --git a/src/trace_processor/sqlite/perfetto_sql_parser.h b/src/trace_processor/sqlite/perfetto_sql_parser.h
index 85622bb..a4a5cc5 100644
--- a/src/trace_processor/sqlite/perfetto_sql_parser.h
+++ b/src/trace_processor/sqlite/perfetto_sql_parser.h
@@ -21,6 +21,7 @@
 #include <variant>
 
 #include "perfetto/ext/base/status_or.h"
+#include "src/trace_processor/sqlite/sql_source.h"
 #include "src/trace_processor/sqlite/sqlite_tokenizer.h"
 
 namespace perfetto {
@@ -41,18 +42,14 @@
   // Indicates that the specified SQLite SQL was extracted directly from a
   // PerfettoSQL statement and should be directly executed with SQLite.
   struct SqliteSql {
-    std::string_view sql;
-    uint32_t global_pos;
-
-    bool operator==(const SqliteSql& o) const {
-      return sql == o.sql && global_pos == o.global_pos;
-    }
+    SqlSource sql;
+    bool operator==(const SqliteSql& c) const { return sql == c.sql; }
   };
   using Statement = std::variant<SqliteSql>;
 
   // Creates a new SQL parser with the a block of PerfettoSQL statements.
   // Concretely, the passed string can contain >1 statement.
-  explicit PerfettoSqlParser(const char* sql);
+  explicit PerfettoSqlParser(SqlSource);
 
   // Attempts to parse to the next statement in the SQL. Returns true if
   // a statement was successfully parsed and false if EOF was reached or the
@@ -74,8 +71,12 @@
   const base::Status& status() const { return status_; }
 
  private:
+  // This cannot be moved because we keep pointers into |sql_| in |tokenizer_|.
+  PerfettoSqlParser(PerfettoSqlParser&&) = delete;
+  PerfettoSqlParser& operator=(PerfettoSqlParser&&) = delete;
+
+  SqlSource sql_;
   SqliteTokenizer tokenizer_;
-  const char* start_ = nullptr;
   base::Status status_;
   std::optional<Statement> statement_;
 };
diff --git a/src/trace_processor/sqlite/perfetto_sql_parser_unittest.cc b/src/trace_processor/sqlite/perfetto_sql_parser_unittest.cc
index 014b5b2..7bbf933 100644
--- a/src/trace_processor/sqlite/perfetto_sql_parser_unittest.cc
+++ b/src/trace_processor/sqlite/perfetto_sql_parser_unittest.cc
@@ -21,6 +21,7 @@
 
 #include "perfetto/base/logging.h"
 #include "perfetto/ext/base/status_or.h"
+#include "src/trace_processor/sqlite/sql_source.h"
 #include "test/gtest_and_gmock.h"
 
 namespace perfetto {
@@ -33,7 +34,7 @@
 class PerfettoSqlParserTest : public ::testing::Test {
  protected:
   base::StatusOr<std::vector<PerfettoSqlParser::Statement>> Parse(
-      const char* sql) {
+      SqlSource sql) {
     PerfettoSqlParser parser(sql);
     std::vector<PerfettoSqlParser::Statement> results;
     while (parser.Next()) {
@@ -47,25 +48,24 @@
 };
 
 TEST_F(PerfettoSqlParserTest, Empty) {
-  ASSERT_THAT(*Parse(""), testing::IsEmpty());
+  ASSERT_THAT(*Parse(SqlSource::FromExecuteQuery("")), testing::IsEmpty());
 }
 
 TEST_F(PerfettoSqlParserTest, SemiColonTerminatedStatement) {
-  static constexpr char kSql[] = "SELECT * FROM slice;";
-  ASSERT_THAT(*Parse(kSql), testing::ElementsAre(SqliteSql{kSql, 0}));
+  auto res = SqlSource::FromExecuteQuery("SELECT * FROM slice;");
+  ASSERT_THAT(*Parse(res), testing::ElementsAre(SqliteSql{res}));
 }
 
 TEST_F(PerfettoSqlParserTest, MultipleStmts) {
-  static constexpr char kSql[] = "SELECT * FROM slice; SELECT * FROM s";
-  ASSERT_THAT(*Parse(kSql),
-              testing::ElementsAre(SqliteSql{"SELECT * FROM slice;", 0},
-                                   SqliteSql{"SELECT * FROM s", 21}));
+  auto res =
+      SqlSource::FromExecuteQuery("SELECT * FROM slice; SELECT * FROM s");
+  ASSERT_THAT(*Parse(res), testing::ElementsAre(SqliteSql{res.Substr(0, 20)},
+                                                SqliteSql{res.Substr(21, 15)}));
 }
 
 TEST_F(PerfettoSqlParserTest, IgnoreOnlySpace) {
-  static constexpr char kSql[] = " ; SELECT * FROM s; ; ;";
-  ASSERT_THAT(*Parse(kSql),
-              testing::ElementsAre(SqliteSql{"SELECT * FROM s;", 3}));
+  auto res = SqlSource::FromExecuteQuery(" ; SELECT * FROM s; ; ;");
+  ASSERT_THAT(*Parse(res), testing::ElementsAre(SqliteSql{res.Substr(3, 16)}));
 }
 
 }  // namespace
diff --git a/src/trace_processor/sqlite/sql_source.cc b/src/trace_processor/sqlite/sql_source.cc
new file mode 100644
index 0000000..0cabc1d
--- /dev/null
+++ b/src/trace_processor/sqlite/sql_source.cc
@@ -0,0 +1,138 @@
+/*
+ * Copyright (C) 2023 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/sqlite/sql_source.h"
+
+#include <algorithm>
+#include <iterator>
+#include <string>
+#include <string_view>
+#include <utility>
+
+#include "perfetto/base/logging.h"
+#include "perfetto/ext/base/string_utils.h"
+
+namespace perfetto {
+namespace trace_processor {
+
+namespace {
+
+std::pair<uint32_t, uint32_t> UpdateLineAndColumnForOffset(
+    const std::string& sql,
+    uint32_t line,
+    uint32_t column,
+    uint32_t offset) {
+  if (offset == 0) {
+    return std::make_pair(line, column);
+  }
+
+  const char* new_start = sql.c_str() + offset;
+  size_t prev_nl = sql.rfind('\n', offset - 1);
+  ssize_t nl_count = std::count(sql.c_str(), new_start, '\n');
+  PERFETTO_DCHECK((nl_count == 0) == (prev_nl == std::string_view::npos));
+
+  if (prev_nl == std::string::npos) {
+    return std::make_pair(line + static_cast<uint32_t>(nl_count),
+                          column + static_cast<uint32_t>(offset));
+  }
+
+  ssize_t new_column = std::distance(sql.c_str() + prev_nl, new_start);
+  return std::make_pair(line + static_cast<uint32_t>(nl_count),
+                        static_cast<uint32_t>(new_column));
+}
+
+}  // namespace
+
+SqlSource::SqlSource(std::string sql,
+                     std::string name,
+                     bool include_traceback_header,
+                     uint32_t line,
+                     uint32_t col)
+    : sql_(std::move(sql)),
+      name_(std::move(name)),
+      include_traceback_header_(include_traceback_header),
+      line_(line),
+      col_(col) {}
+
+SqlSource SqlSource::FromExecuteQuery(std::string sql) {
+  return SqlSource(std::move(sql), "File \"stdin\"", true, 1, 1);
+}
+
+SqlSource SqlSource::FromMetric(std::string sql, const std::string& name) {
+  return SqlSource(std::move(sql), "Metric \"" + name + "\"", true, 1, 1);
+}
+
+SqlSource SqlSource::FromFunction(std::string sql, const std::string& name) {
+  return SqlSource(std::move(sql), "Function \"" + name + "\"", false, 1, 1);
+}
+
+SqlSource SqlSource::FromMetricFile(std::string sql, const std::string& name) {
+  return SqlSource(std::move(sql), "Metric file \"" + name + "\"", false, 1, 1);
+}
+
+SqlSource SqlSource::FromModuleImport(std::string sql,
+                                      const std::string& module) {
+  return SqlSource(std::move(sql), "Module import \"" + module + "\"", false, 1,
+                   1);
+}
+
+SqlSource SqlSource::FromSpanJoin(std::string sql, const std::string& name) {
+  return SqlSource(std::move(sql), "Span Join Table \"" + name + "\"", false, 1,
+                   1);
+}
+
+SqlSource SqlSource::Substr(uint32_t offset, uint32_t len) {
+  auto line_and_col = UpdateLineAndColumnForOffset(sql_, line_, col_, offset);
+  return SqlSource(sql_.substr(offset, len), name_, include_traceback_header_,
+                   line_and_col.first, line_and_col.second);
+}
+
+std::string SqlSource::AsTracebackFrame(
+    std::optional<uint32_t> opt_offset) const {
+  uint32_t offset = opt_offset.value_or(0);
+
+  size_t start_idx = offset - std::min<size_t>(128ul, offset);
+  if (offset > 0) {
+    size_t prev_nl = sql_.rfind('\n', offset - 1);
+    if (prev_nl != std::string::npos) {
+      start_idx = std::max(prev_nl + 1, start_idx);
+    }
+  }
+
+  size_t end_idx = std::min<size_t>(offset + 128ul, sql_.size());
+  size_t next_nl = sql_.find('\n', offset);
+  if (next_nl != std::string::npos) {
+    end_idx = std::min(next_nl, end_idx);
+  }
+  size_t caret_pos = offset - start_idx;
+
+  std::string header;
+  if (include_traceback_header_) {
+    header = "Traceback (most recent call last):\n";
+  }
+
+  auto line_and_col = UpdateLineAndColumnForOffset(sql_, line_, col_, offset);
+  std::string sql_segment = sql_.substr(start_idx, end_idx - start_idx);
+  std::string caret = std::string(caret_pos, ' ') + "^";
+  base::StackString<1024> str("%s  %s line %u col %u\n    %s\n    %s\n",
+                              header.c_str(), name_.c_str(), line_and_col.first,
+                              line_and_col.second, sql_segment.c_str(),
+                              caret.c_str());
+  return str.ToStdString();
+}
+
+}  // namespace trace_processor
+}  // namespace perfetto
diff --git a/src/trace_processor/sqlite/sql_source.h b/src/trace_processor/sqlite/sql_source.h
new file mode 100644
index 0000000..16c0491
--- /dev/null
+++ b/src/trace_processor/sqlite/sql_source.h
@@ -0,0 +1,100 @@
+/*
+ * Copyright (C) 2018 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_SQLITE_SQL_SOURCE_H_
+#define SRC_TRACE_PROCESSOR_SQLITE_SQL_SOURCE_H_
+
+#include <optional>
+#include <string>
+#include <string_view>
+#include <tuple>
+
+namespace perfetto {
+namespace trace_processor {
+
+// An SQL string which retains knowledge of the source of the SQL (i.e. stdlib
+// module, ExecuteQuery etc).
+//
+// The reason this class exists is to allow much better error messages as we
+// can not only render of the snippet of SQL which is failing but also point
+// to the correct line number in the context of the whole SQL file.
+class SqlSource {
+ public:
+  // Creates a SqlSource instance wrapping SQL passed to
+  // |TraceProcessor::ExecuteQuery|.
+  static SqlSource FromExecuteQuery(std::string sql);
+
+  // Creates a SqlSource instance wrapping SQL executed when running a metric.
+  static SqlSource FromMetric(std::string sql, const std::string& metric_file);
+
+  // Creates a SqlSource instance wrapping SQL executed when running a metric
+  // file (i.e. with RUN_METRIC).
+  static SqlSource FromMetricFile(std::string sql,
+                                  const std::string& metric_file);
+
+  // Creates a SqlSource instance wrapping SQL executed when importing a module.
+  static SqlSource FromModuleImport(std::string sql, const std::string& module);
+
+  // Creates a SqlSource instance wrapping SQL executed when running a function.
+  static SqlSource FromFunction(std::string sql, const std::string& function);
+
+  // Creates a SqlSource instance wrapping SQL executed when executing a SPAN
+  // JOIN.
+  static SqlSource FromSpanJoin(std::string sql,
+                                const std::string& span_join_table);
+
+  // Creates a SqlSource instance with the SQL taken as a substring starting at
+  // |offset| with |len| characters.
+  SqlSource Substr(uint32_t offset, uint32_t len);
+
+  // Returns the this SqlSource instance as a string which can be appended as a
+  // "traceback" frame to an error message. Callers can pass an optional
+  // |offset| parameter which indicates the exact location of the error in the
+  // SQL string.
+  //
+  // Specifically, this string will include:
+  //  a) context about the source of the SQL
+  //  b) line and column number of the error
+  //  c) a snippet of the SQL and a caret (^) character pointing to the location
+  //     of the error.
+  std::string AsTracebackFrame(std::optional<uint32_t> offset) const;
+
+  // Returns the SQL backing this SqlSource instance;
+  const std::string& sql() const { return sql_; }
+
+  bool operator==(const SqlSource& other) const {
+    return std::tie(sql_, line_, col_) ==
+           std::tie(other.sql_, other.line_, other.col_);
+  }
+
+ private:
+  SqlSource(std::string sql,
+            std::string name,
+            bool include_traceback_header,
+            uint32_t line,
+            uint32_t col);
+
+  std::string sql_;
+  std::string name_;
+  bool include_traceback_header_ = false;
+  uint32_t line_ = 1;
+  uint32_t col_ = 1;
+};
+
+}  // namespace trace_processor
+}  // namespace perfetto
+
+#endif  // SRC_TRACE_PROCESSOR_SQLITE_SQL_SOURCE_H_
diff --git a/src/trace_processor/sqlite/sqlite_engine.cc b/src/trace_processor/sqlite/sqlite_engine.cc
index 8414193..a501efa 100644
--- a/src/trace_processor/sqlite/sqlite_engine.cc
+++ b/src/trace_processor/sqlite/sqlite_engine.cc
@@ -16,12 +16,16 @@
 
 #include "src/trace_processor/sqlite/sqlite_engine.h"
 
+#include <optional>
 #include <utility>
 
 #include "perfetto/base/status.h"
 #include "src/trace_processor/sqlite/db_sqlite_table.h"
 #include "src/trace_processor/sqlite/query_cache.h"
+#include "src/trace_processor/sqlite/scoped_db.h"
+#include "src/trace_processor/sqlite/sql_source.h"
 #include "src/trace_processor/sqlite/sqlite_table.h"
+#include "src/trace_processor/sqlite/sqlite_utils.h"
 
 // In Android and Chromium tree builds, we don't have the percentile module.
 // Just don't include it.
@@ -61,6 +65,12 @@
 #endif
 }
 
+std::optional<uint32_t> GetErrorOffsetDb(sqlite3* db) {
+  int offset = sqlite3_error_offset(db);
+  return offset == -1 ? std::nullopt
+                      : std::make_optional(static_cast<uint32_t>(offset));
+}
+
 }  // namespace
 
 SqliteEngine::SqliteEngine() {
@@ -84,6 +94,20 @@
   fn_ctx_.Clear();
 }
 
+base::StatusOr<SqliteEngine::PreparedStatement> SqliteEngine::PrepareStatement(
+    SqlSource sql) {
+  PERFETTO_TP_TRACE(metatrace::Category::QUERY, "QUERY_PREPARE");
+  sqlite3_stmt* raw_stmt = nullptr;
+  int err =
+      sqlite3_prepare_v2(db_.get(), sql.sql().c_str(), -1, &raw_stmt, nullptr);
+  if (err != SQLITE_OK) {
+    const char* errmsg = sqlite3_errmsg(db_.get());
+    std::string frame = sql.AsTracebackFrame(GetErrorOffset());
+    return base::ErrStatus("%s%s", frame.c_str(), errmsg);
+  }
+  return PreparedStatement{ScopedStmt(raw_stmt), std::move(sql)};
+}
+
 base::Status SqliteEngine::RegisterFunction(const char* name,
                                             int argc,
                                             Fn* fn,
@@ -136,5 +160,50 @@
   return res ? *res : nullptr;
 }
 
+std::optional<uint32_t> SqliteEngine::GetErrorOffset() const {
+  return GetErrorOffsetDb(db_.get());
+}
+
+SqliteEngine::PreparedStatement::PreparedStatement(ScopedStmt stmt,
+                                                   SqlSource tagged)
+    : stmt_(std::move(stmt)), sql_source_(std::move(tagged)) {}
+
+bool SqliteEngine::PreparedStatement::Step() {
+  PERFETTO_TP_TRACE(metatrace::Category::TOPLEVEL, "STMT_STEP",
+                    [this](metatrace::Record* record) {
+                      record->AddArg("SQL", expanded_sql());
+                    });
+
+  // Now step once into |cur_stmt| so that when we prepare the next statment
+  // we will have executed any dependent bytecode in this one.
+  int err = sqlite3_step(stmt_.get());
+  if (err == SQLITE_ROW) {
+    return true;
+  }
+  if (err == SQLITE_DONE) {
+    return false;
+  }
+  sqlite3* db = sqlite3_db_handle(stmt_.get());
+  std::string frame = sql_source_.AsTracebackFrame(GetErrorOffsetDb(db));
+  const char* errmsg = sqlite3_errmsg(db);
+  status_ = base::ErrStatus("%s%s", frame.c_str(), errmsg);
+  return false;
+}
+
+bool SqliteEngine::PreparedStatement::IsDone() const {
+  return !sqlite3_stmt_busy(stmt_.get());
+}
+
+const char* SqliteEngine::PreparedStatement::sql() const {
+  return sqlite3_sql(stmt_.get());
+}
+
+const char* SqliteEngine::PreparedStatement::expanded_sql() {
+  if (!expanded_sql_) {
+    expanded_sql_.reset(sqlite3_expanded_sql(stmt_.get()));
+  }
+  return expanded_sql_.get();
+}
+
 }  // namespace trace_processor
 }  // namespace perfetto
diff --git a/src/trace_processor/sqlite/sqlite_engine.h b/src/trace_processor/sqlite/sqlite_engine.h
index 6c5ad10..80cf7a3 100644
--- a/src/trace_processor/sqlite/sqlite_engine.h
+++ b/src/trace_processor/sqlite/sqlite_engine.h
@@ -21,6 +21,7 @@
 #include <stdint.h>
 #include <functional>
 #include <memory>
+#include <optional>
 #include <type_traits>
 
 #include "perfetto/base/status.h"
@@ -31,8 +32,10 @@
 #include "src/trace_processor/prelude/table_functions/table_function.h"
 #include "src/trace_processor/sqlite/query_cache.h"
 #include "src/trace_processor/sqlite/scoped_db.h"
+#include "src/trace_processor/sqlite/sql_source.h"
 #include "src/trace_processor/sqlite/sqlite_table.h"
 #include "src/trace_processor/sqlite/sqlite_utils.h"
+#include "src/trace_processor/tp_metatrace.h"
 
 namespace perfetto {
 namespace trace_processor {
@@ -50,9 +53,35 @@
   using Fn = void(sqlite3_context* ctx, int argc, sqlite3_value** argv);
   using FnCtxDestructor = void(void*);
 
+  // Wrapper class for SQLite's |sqlite3_stmt| struct and associated functions.
+  struct PreparedStatement {
+   public:
+    bool Step();
+    bool IsDone() const;
+
+    const char* sql() const;
+    const char* expanded_sql();
+
+    const base::Status& status() const { return status_; }
+    sqlite3_stmt* sqlite_stmt() const { return stmt_.get(); }
+
+   private:
+    friend class SqliteEngine;
+
+    explicit PreparedStatement(ScopedStmt, SqlSource);
+
+    ScopedStmt stmt_;
+    SqlSource sql_source_;
+    ScopedSqliteString expanded_sql_;
+    base::Status status_;
+  };
+
   SqliteEngine();
   ~SqliteEngine();
 
+  // Prepares a SQLite statement for the given SQL.
+  base::StatusOr<PreparedStatement> PrepareStatement(SqlSource);
+
   // Registers a C++ function to be runnable from SQL.
   base::Status RegisterFunction(const char* name,
                                 int argc,
@@ -94,6 +123,11 @@
     }
   };
 
+  std::optional<uint32_t> GetErrorOffset() const;
+
+  SqliteEngine(SqliteEngine&&) noexcept = delete;
+  SqliteEngine& operator=(SqliteEngine&&) = delete;
+
   base::FlatHashMap<std::string, std::unique_ptr<SqliteTable>> saved_tables_;
   base::FlatHashMap<std::pair<std::string, int>, void*, FnHasher> fn_ctx_;
 
diff --git a/src/trace_processor/sqlite/sqlite_utils.h b/src/trace_processor/sqlite/sqlite_utils.h
index d3ccfc5..5d59e41 100644
--- a/src/trace_processor/sqlite/sqlite_utils.h
+++ b/src/trace_processor/sqlite/sqlite_utils.h
@@ -23,6 +23,7 @@
 #include <cstddef>
 #include <cstring>
 #include <optional>
+#include <string>
 #include <utility>
 
 #include "perfetto/base/logging.h"
@@ -144,76 +145,6 @@
   return ScopedSqliteString(sqlite3_expanded_sql(stmt));
 }
 
-inline base::Status FormatErrorMessage(base::StringView sql,
-                                       sqlite3* db,
-                                       int error_code) {
-  uint32_t offset = static_cast<uint32_t>(sqlite3_error_offset(db));
-
-  auto error_opt = FindLineWithOffset(sql, offset);
-
-  if (!error_opt.has_value()) {
-    return base::ErrStatus("Error: %s (errcode: %d)", sqlite3_errmsg(db),
-                           error_code);
-  }
-
-  auto error = error_opt.value();
-
-  return base::ErrStatus(
-      "Error in line:%u, col: %u.\n"
-      "%s\n"
-      "%s^\n"
-      "%s (errcode: %d)",
-      error.line_num, error.line_offset + 1, error.line.ToStdString().c_str(),
-      std::string(error.line_offset, ' ').c_str(), sqlite3_errmsg(db),
-      error_code);
-}
-
-inline base::Status FormatErrorMessage(sqlite3_stmt* stmt,
-                                       std::optional<base::StringView> sql,
-                                       sqlite3* db,
-                                       int error_code) {
-  if (stmt) {
-    auto expanded_sql = ExpandedSqlForStmt(stmt);
-    PERFETTO_CHECK(expanded_sql);
-    return FormatErrorMessage(expanded_sql.get(), db, error_code);
-  }
-  PERFETTO_CHECK(sql.has_value());
-  return FormatErrorMessage(sql.value(), db, error_code);
-}
-
-inline base::Status PrepareStmt(sqlite3* db,
-                                const char* sql,
-                                ScopedStmt* stmt,
-                                const char** tail) {
-  sqlite3_stmt* raw_stmt = nullptr;
-  int err = sqlite3_prepare_v2(db, sql, -1, &raw_stmt, tail);
-  stmt->reset(raw_stmt);
-  if (err != SQLITE_OK)
-    return base::ErrStatus("%s", FormatErrorMessage(sql, db, err).c_message());
-  return base::OkStatus();
-}
-
-inline bool IsStmtDone(sqlite3_stmt* stmt) {
-  return !sqlite3_stmt_busy(stmt);
-}
-
-inline base::Status StepStmtUntilDone(sqlite3_stmt* stmt) {
-  PERFETTO_DCHECK(stmt);
-
-  if (IsStmtDone(stmt))
-    return base::OkStatus();
-
-  int err;
-  for (err = sqlite3_step(stmt); err == SQLITE_ROW; err = sqlite3_step(stmt)) {
-  }
-  if (err != SQLITE_DONE) {
-    auto db = sqlite3_db_handle(stmt);
-    return base::ErrStatus(
-        "%s", FormatErrorMessage(stmt, std::nullopt, db, err).c_message());
-  }
-  return base::OkStatus();
-}
-
 inline void SetSqliteError(sqlite3_context* ctx, const base::Status& status) {
   PERFETTO_CHECK(!status.ok());
   sqlite3_result_error(ctx, status.c_message(), -1);
diff --git a/src/trace_processor/trace_database_integrationtest.cc b/src/trace_processor/trace_database_integrationtest.cc
index b675c80..3be0a12 100644
--- a/src/trace_processor/trace_database_integrationtest.cc
+++ b/src/trace_processor/trace_database_integrationtest.cc
@@ -15,6 +15,7 @@
  */
 
 #include <algorithm>
+#include <cstdio>
 #include <map>
 #include <optional>
 #include <random>
@@ -22,6 +23,7 @@
 
 #include "perfetto/base/logging.h"
 #include "perfetto/ext/base/scoped_file.h"
+#include "perfetto/trace_processor/basic_types.h"
 #include "perfetto/trace_processor/trace_processor.h"
 #include "protos/perfetto/common/descriptor.pbzero.h"
 #include "protos/perfetto/trace_processor/trace_processor.pbzero.h"
@@ -486,6 +488,60 @@
   EXPECT_STREQ(it.Get(0).string_value, "123e4567-e89b-12d3-a456-426655443322");
 }
 
+TEST_F(TraceProcessorIntegrationTest, ErrorMessageExecuteQuery) {
+  auto it = Query("select t from slice");
+  ASSERT_FALSE(it.Next());
+  ASSERT_FALSE(it.Status().ok());
+
+  ASSERT_THAT(it.Status().message(),
+              testing::Eq(R"(Traceback (most recent call last):
+  File "stdin" line 1 col 8
+    select t from slice
+           ^
+no such column: t)"));
+}
+
+TEST_F(TraceProcessorIntegrationTest, ErrorMessageMetricFile) {
+  ASSERT_TRUE(
+      Processor()->RegisterMetric("foo/bar.sql", "select t from slice").ok());
+
+  auto it = Query("select RUN_METRIC('foo/bar.sql');");
+  ASSERT_FALSE(it.Next());
+  ASSERT_FALSE(it.Status().ok());
+
+  ASSERT_EQ(it.Status().message(),
+            R"(Traceback (most recent call last):
+  File "stdin" line 1 col 1
+    select RUN_METRIC('foo/bar.sql');
+    ^
+  Metric file "foo/bar.sql" line 1 col 8
+    select t from slice
+           ^
+no such column: t)");
+}
+
+TEST_F(TraceProcessorIntegrationTest, ErrorMessageModule) {
+  SqlModule module;
+  module.name = "foo";
+  module.files.push_back(std::make_pair("foo.bar", "select t from slice"));
+
+  ASSERT_TRUE(Processor()->RegisterSqlModule(module).ok());
+
+  auto it = Query("select IMPORT('foo.bar');");
+  ASSERT_FALSE(it.Next());
+  ASSERT_FALSE(it.Status().ok());
+
+  ASSERT_EQ(it.Status().message(),
+            R"(Traceback (most recent call last):
+  File "stdin" line 1 col 1
+    select IMPORT('foo.bar');
+    ^
+  Module import "foo.bar" line 1 col 8
+    select t from slice
+           ^
+no such column: t)");
+}
+
 }  // namespace
 }  // namespace trace_processor
 }  // namespace perfetto
diff --git a/src/trace_processor/trace_processor_impl.cc b/src/trace_processor/trace_processor_impl.cc
index 2ce3871..c28e373 100644
--- a/src/trace_processor/trace_processor_impl.cc
+++ b/src/trace_processor/trace_processor_impl.cc
@@ -76,6 +76,7 @@
 #include "src/trace_processor/prelude/tables_views/tables_views.h"
 #include "src/trace_processor/sqlite/perfetto_sql_engine.h"
 #include "src/trace_processor/sqlite/scoped_db.h"
+#include "src/trace_processor/sqlite/sql_source.h"
 #include "src/trace_processor/sqlite/sql_stats_table.h"
 #include "src/trace_processor/sqlite/sqlite_table.h"
 #include "src/trace_processor/sqlite/sqlite_utils.h"
@@ -276,7 +277,7 @@
   RegisterFunction<metrics::RunMetric>(
       engine, "RUN_METRIC", -1,
       std::unique_ptr<metrics::RunMetric::Context>(
-          new metrics::RunMetric::Context{tp, sql_metrics}));
+          new metrics::RunMetric::Context{engine, sql_metrics}));
 
   // TODO(lalitm): migrate this over to using RegisterFunction once aggregate
   // functions are supported.
@@ -418,16 +419,13 @@
                                 context_.clock_converter.get());
   RegisterFunction<ToTimecode>(&engine_, "TO_TIMECODE", 1);
   RegisterFunction<CreateFunction>(&engine_, "CREATE_FUNCTION", 3, &engine_);
+  RegisterFunction<CreateViewFunction>(&engine_, "CREATE_VIEW_FUNCTION", 3,
+                                       &engine_);
   RegisterFunction<ExperimentalMemoize>(&engine_, "EXPERIMENTAL_MEMOIZE", 1,
                                         &engine_);
-  RegisterFunction<CreateViewFunction>(
-      &engine_, "CREATE_VIEW_FUNCTION", 3,
-      std::unique_ptr<CreateViewFunction::Context>(
-          new CreateViewFunction::Context{engine_.sqlite_engine()->db()}));
-  RegisterFunction<Import>(
-      &engine_, "IMPORT", 1,
-      std::unique_ptr<Import::Context>(new Import::Context{
-          engine_.sqlite_engine()->db(), this, &sql_modules_}));
+  RegisterFunction<Import>(&engine_, "IMPORT", 1,
+                           std::unique_ptr<Import::Context>(
+                               new Import::Context{&engine_, &sql_modules_}));
   RegisterFunction<ToFtrace>(
       &engine_, "TO_FTRACE", 1,
       std::unique_ptr<ToFtrace::Context>(new ToFtrace::Context{
@@ -465,16 +463,16 @@
 
   // Operator tables.
   engine_.sqlite_engine()->RegisterVirtualTableModule<SpanJoinOperatorTable>(
-      "span_join", storage, SqliteTable::TableType::kExplicitCreate, false);
+      "span_join", &engine_, SqliteTable::TableType::kExplicitCreate, false);
   engine_.sqlite_engine()->RegisterVirtualTableModule<SpanJoinOperatorTable>(
-      "span_left_join", storage, SqliteTable::TableType::kExplicitCreate,
+      "span_left_join", &engine_, SqliteTable::TableType::kExplicitCreate,
       false);
   engine_.sqlite_engine()->RegisterVirtualTableModule<SpanJoinOperatorTable>(
-      "span_outer_join", storage, SqliteTable::TableType::kExplicitCreate,
+      "span_outer_join", &engine_, SqliteTable::TableType::kExplicitCreate,
       false);
   engine_.sqlite_engine()->RegisterVirtualTableModule<WindowOperatorTable>(
       "window", storage, SqliteTable::TableType::kExplicitCreate, true);
-  RegisterCreateViewFunctionModule(engine_.sqlite_engine());
+  RegisterCreateViewFunctionModule(&engine_);
 
   // Initalize the tables and views in the prelude.
   InitializePreludeTablesViews(engine_.sqlite_engine()->db());
@@ -724,7 +722,8 @@
           sql, base::GetWallTimeNs().count());
 
   base::StatusOr<PerfettoSqlEngine::ExecutionResult> result =
-      engine_.ExecuteUntilLastStatement(sql);
+      engine_.ExecuteUntilLastStatement(
+          SqlSource::FromExecuteQuery(sql.c_str()));
   std::unique_ptr<IteratorImpl> impl(
       new IteratorImpl(this, std::move(result), sql_stats_row));
   return Iterator(std::move(impl));
@@ -773,21 +772,13 @@
 
 base::Status TraceProcessorImpl::RegisterMetric(const std::string& path,
                                                 const std::string& sql) {
-  std::string stripped_sql;
-  for (base::StringSplitter sp(sql, '\n'); sp.Next();) {
-    if (strncmp(sp.cur_token(), "--", 2) != 0) {
-      stripped_sql.append(sp.cur_token());
-      stripped_sql.push_back('\n');
-    }
-  }
-
   // Check if the metric with the given path already exists and if it does,
   // just update the SQL associated with it.
   auto it = std::find_if(
       sql_metrics_.begin(), sql_metrics_.end(),
       [&path](const metrics::SqlMetricFile& m) { return m.path == path; });
   if (it != sql_metrics_.end()) {
-    it->sql = stripped_sql;
+    it->sql = sql;
     return base::OkStatus();
   }
 
@@ -803,7 +794,7 @@
 
   metrics::SqlMetricFile metric;
   metric.path = path;
-  metric.sql = stripped_sql;
+  metric.sql = sql;
 
   if (IsRootMetricField(no_ext_name)) {
     metric.proto_field_name = no_ext_name;
@@ -869,7 +860,7 @@
     return base::Status("Root metrics proto descriptor not found");
 
   const auto& root_descriptor = pool_.descriptors()[opt_idx.value()];
-  return metrics::ComputeMetrics(this, metric_names, sql_metrics_, pool_,
+  return metrics::ComputeMetrics(&engine_, metric_names, sql_metrics_, pool_,
                                  root_descriptor, metrics_proto);
 }
 
diff --git a/src/trace_processor/trace_processor_shell.cc b/src/trace_processor/trace_processor_shell.cc
index 5d86b7c..7daec93 100644
--- a/src/trace_processor/trace_processor_shell.cc
+++ b/src/trace_processor/trace_processor_shell.cc
@@ -271,7 +271,7 @@
 
   base::Status status = attach_it.Status();
   if (!status.ok())
-    return base::ErrStatus("SQLite error: %s", status.c_message());
+    return base::ErrStatus("%s", status.c_message());
 
   // Export real and virtual tables.
   auto tables_it = g_tp->ExecuteQuery(
@@ -289,11 +289,11 @@
 
     status = export_it.Status();
     if (!status.ok())
-      return base::ErrStatus("SQLite error: %s", status.c_message());
+      return base::ErrStatus("%s", status.c_message());
   }
   status = tables_it.Status();
   if (!status.ok())
-    return base::ErrStatus("SQLite error: %s", status.c_message());
+    return base::ErrStatus("%s", status.c_message());
 
   // Export views.
   auto views_it =
@@ -313,18 +313,18 @@
 
     status = export_it.Status();
     if (!status.ok())
-      return base::ErrStatus("SQLite error: %s", status.c_message());
+      return base::ErrStatus("%s", status.c_message());
   }
   status = views_it.Status();
   if (!status.ok())
-    return base::ErrStatus("SQLite error: %s", status.c_message());
+    return base::ErrStatus("%s", status.c_message());
 
   auto detach_it = g_tp->ExecuteQuery("DETACH DATABASE perfetto_export");
   bool detach_has_more = attach_it.Next();
   PERFETTO_DCHECK(!detach_has_more);
   status = detach_it.Status();
   return status.ok() ? base::OkStatus()
-                     : base::ErrStatus("SQLite error: %s", status.c_message());
+                     : base::ErrStatus("%s", status.c_message());
 }
 
 class ErrorPrinter : public google::protobuf::io::ErrorCollector {
@@ -413,8 +413,7 @@
     base::Status status =
         g_tp->ComputeMetricText(metric_names, TraceProcessor::kProtoText, &out);
     if (!status.ok()) {
-      return base::ErrStatus("Error when computing metrics: %s",
-                             status.c_message());
+      return status;
     }
     out += '\n';
     fwrite(out.c_str(), sizeof(char), out.size(), stdout);
@@ -422,12 +421,7 @@
   }
 
   std::vector<uint8_t> metric_result;
-  base::Status status = g_tp->ComputeMetric(metric_names, &metric_result);
-  if (!status.ok()) {
-    return base::ErrStatus("Error when computing metrics: %s",
-                           status.c_message());
-  }
-
+  RETURN_IF_ERROR(g_tp->ComputeMetric(metric_names, &metric_result));
   switch (format) {
     case OutputFormat::kJson: {
       // TODO(b/182165266): Handle this using ComputeMetricText.
@@ -517,7 +511,7 @@
 
   base::Status status = it->Status();
   if (!status.ok()) {
-    PERFETTO_ELOG("SQLite error: %s", status.c_message());
+    fprintf(stderr, "%s\n", status.c_message());
   }
   printf("\nQuery executed in %.3f ms\n\n",
          static_cast<double>((t_end - t_start).count()) / 1E6);
@@ -572,7 +566,7 @@
 
 base::Status RunQueriesAndPrintResult(const std::string& sql_query,
                                       FILE* output) {
-  PERFETTO_ILOG("Executing query: %s", sql_query.c_str());
+  PERFETTO_DLOG("Executing query: %s", sql_query.c_str());
   auto query_start = std::chrono::steady_clock::now();
 
   auto it = g_tp->ExecuteQuery(sql_query);
@@ -1097,7 +1091,9 @@
 base::Status RunQueries(const std::string& query_file_path,
                         bool expect_output) {
   std::string queries;
-  base::ReadFile(query_file_path.c_str(), &queries);
+  if (!base::ReadFile(query_file_path.c_str(), &queries)) {
+    return base::ErrStatus("Unable to read file %s", query_file_path.c_str());
+  }
 
   base::Status status;
   if (expect_output) {
@@ -1106,8 +1102,7 @@
     status = RunQueriesWithoutOutput(queries);
   }
   if (!status.ok()) {
-    return base::ErrStatus("Encountered error while running queries: %s",
-                           status.c_message());
+    return base::ErrStatus("%s", status.c_message());
   }
   return base::OkStatus();
 }
@@ -1501,7 +1496,7 @@
         base::Status status =
             RunMetrics(options.metrics, options.metric_format, *options.pool);
         if (!status.ok()) {
-          PERFETTO_ELOG("%s", status.c_message());
+          fprintf(stderr, "%s\n", status.c_message());
         }
       } else {
         PrintShellUsage();
@@ -1724,7 +1719,7 @@
 int main(int argc, char** argv) {
   auto status = perfetto::trace_processor::TraceProcessorMain(argc, argv);
   if (!status.ok()) {
-    PERFETTO_ELOG("%s", status.c_message());
+    fprintf(stderr, "%s\n", status.c_message());
     return 1;
   }
   return 0;
diff --git a/tools/gen_amalgamated_sql.py b/tools/gen_amalgamated_sql.py
index a726002..a3621f7 100755
--- a/tools/gen_amalgamated_sql.py
+++ b/tools/gen_amalgamated_sql.py
@@ -21,7 +21,7 @@
 # as a string constant to allow trace processor to exectue the metrics.
 
 REPLACEMENT_HEADER = '''/*
- * Copyright (C) 2022 The Android Open Source Project
+ * Copyright (C) 2023 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.
@@ -38,7 +38,7 @@
 
 /*
  *******************************************************************************
- * AUTOGENERATED BY tools/gen_merged_sql_metrics - DO NOT EDIT
+ * AUTOGENERATED BY tools/gen_amalgamated_sql.py - DO NOT EDIT
  *******************************************************************************
  */
 
@@ -106,9 +106,7 @@
       # and ends up with a bunch of ../ prefixing the path: disallow any ../
       # as this should never be a valid in our C++ output.
       assert '../' not in relpath
-
-      sql_outputs[relpath] = "".join(
-          x.lstrip() for x in f.readlines() if not x.lstrip().startswith('--'))
+      sql_outputs[relpath] = f.read()
 
   with open(args.cpp_out, 'w+') as output:
     output.write(REPLACEMENT_HEADER)