diff --git a/Makefile b/Makefile
index b8745e57dc..8b12df9228 100644
--- a/Makefile
+++ b/Makefile
@@ -50,6 +50,12 @@ mysqlsh:
 proto:
 	buf generate
 
+proto-engine-plugin:
+	protoc -I. \
+		--go_out=. --go_opt=module=github.com/sqlc-dev/sqlc --go_opt=Mprotos/engine/engine.proto=github.com/sqlc-dev/sqlc/pkg/engine \
+		--go-grpc_out=. --go-grpc_opt=module=github.com/sqlc-dev/sqlc --go-grpc_opt=Mprotos/engine/engine.proto=github.com/sqlc-dev/sqlc/pkg/engine \
+		protos/engine/engine.proto
+
 remote-proto:
 	protoc \
 		--go_out=. --go_opt="Minternal/remote/gen.proto=github.com/sqlc-dev/sqlc/internal/remote" --go_opt=module=github.com/sqlc-dev/sqlc \
diff --git a/README.md b/README.md
index 43fed122b9..f2ae62495c 100644
--- a/README.md
+++ b/README.md
@@ -29,6 +29,14 @@ Check out [an interactive example](https://play.sqlc.dev/) to see it in action,
 
 Additional languages can be added via [plugins](https://docs.sqlc.dev/en/latest/reference/language-support.html#community-language-support).
 
+## Supported database engines
+
+- PostgreSQL
+- MySQL
+- SQLite
+
+Additional database engines can be added via [engine plugins](https://docs.sqlc.dev/en/latest/guides/engine-plugins.html).
+
 ## Sponsors
 
 Development is possible thanks to our sponsors. If you would like to support sqlc,
diff --git a/docs/guides/engine-plugins.md b/docs/guides/engine-plugins.md
new file mode 100644
index 0000000000..9acd889da9
--- /dev/null
+++ b/docs/guides/engine-plugins.md
@@ -0,0 +1,222 @@
+# External Database Engines (Engine Plugins)
+
+Engine plugins let you use sqlc with databases that are not built-in. You can add support for other SQL-compatible systems (e.g. CockroachDB, TiDB, or custom engines) by implementing a small external program that parses SQL and returns parameters and result columns.
+
+## Why use an engine plugin?
+
+- Use sqlc with a database that doesn't have native support.
+- Reuse an existing SQL parser or dialect in a separate binary.
+- Keep engine-specific logic outside the sqlc core.
+
+Data returned by the engine plugin (SQL text, parameters, columns) is passed through to [codegen plugins](plugins.md) without an extra compiler/AST step. The plugin is the single place that defines how queries are interpreted for that engine.
+
+**Limitation:** `sqlc vet` does not support plugin engines. Use vet only with built-in engines (postgresql, mysql, sqlite).
+
+## Overview
+
+An engine plugin is an external process that implements one RPC:
+
+- **Parse** — accepts the **entire contents** of one query file (e.g. `query.sql`) and either schema SQL or connection parameters; returns **one Statement per query block** in that file (each with sql, parameters, columns, and name/cmd).
+
+Process plugins (e.g. written in Go) talk to sqlc over **stdin/stdout** using **Protocol Buffers**. The protocol is defined in `protos/engine/engine.proto`.
+
+## Compatibility
+
+For Go plugins, compatibility is enforced at **compile time** by importing the engine package:
+
+```go
+import "github.com/sqlc-dev/sqlc/pkg/engine"
+```
+
+- If the plugin builds, it matches this version of the engine API.
+- If the API changes in a breaking way, the plugin stops compiling until it's updated.
+
+No version handshake is required; the proto schema defines the contract.
+
+## Configuration
+
+### sqlc.yaml
+
+```yaml
+version: "2"
+
+engines:
+  - name: external-db
+    process:
+      # Executable and optional arguments (e.g. --dont-open-wildcard-star).
+      # First token is the command; the rest are passed to the plugin before the RPC method name.
+      cmd: sqlc-engine-external-db --dont-open-wildcard-star
+    env:
+      - EXTERNAL_DB_DSN
+
+sql:
+  - engine: external-db
+    schema: "schema.sql"
+    queries: "queries.sql"
+    codegen:
+      - plugin: go
+        out: db
+```
+
+### Engine options
+
+| Field | Description |
+|-------|-------------|
+| `name` | Engine name used in `sql[].engine` |
+| `process.cmd` | Command to run: executable path and optional arguments (e.g. `sqlc-engine-external-db --dont-open-wildcard-star`). First token is the executable; remaining tokens are passed as arguments before the RPC method. |
+| `env` | Environment variable names passed to the plugin |
+
+Each engine must define either `process` (with `cmd`) or `wasm` (with `url` and `sha256`). See [Configuration reference](../reference/config.md) for the full `engines` schema.
+
+### How sqlc finds the process plugin
+
+For an engine with `process.cmd`, sqlc resolves and runs the plugin as follows:
+
+1. **Command parsing** — `process.cmd` is split on whitespace. The first token is the executable; any further tokens are passed as arguments, and sqlc appends the RPC method name (`parse`) when invoking the plugin.
+
+2. **Executable lookup** — The first token is resolved the same way as in the shell:
+   - If it contains a path separator (e.g. `/usr/bin/sqlc-engine-external-db` or `./bin/sqlc-engine-external-db`), it is treated as a path. Absolute paths are used as-is; relative paths are taken relative to the **current working directory of the process running sqlc**.
+   - If it has no path separator, the executable is looked up in the **PATH** of the process running sqlc. The plugin binary must be on PATH (e.g. after `go install` or adding its directory to PATH) or `process.cmd` must be an absolute path.
+
+3. **Working directory** — The plugin process is started with its working directory set to the **directory containing the sqlc config file**. That directory is used for resolving relative paths inside the plugin, not for resolving `process.cmd` itself.
+
+If the executable cannot be found or `process.cmd` is empty, sqlc reports an error and refers to this documentation.
+
+## Implementing an engine plugin (Go)
+
+### 1. Dependencies and entrypoint
+
+```go
+package main
+
+import "github.com/sqlc-dev/sqlc/pkg/engine"
+
+func main() {
+    engine.Run(engine.Handler{
+        PluginName:    "external-db",
+        PluginVersion: "1.0.0",
+        Parse:         handleParse,
+    })
+}
+```
+
+The engine API exposes only **Parse**. There are no separate methods for catalog, keywords, comment syntax, or dialect.
+
+### 2. Parse
+
+sqlc calls Parse **once per query file** (e.g. once for `query.sql`). The plugin receives the full file contents and returns one **Statement** per query block in that file. sqlc then passes each statement to the codegen plugin as a separate query.
+
+**Request**
+
+- `sql` — The **entire contents** of one query file (all query blocks, with `-- name: X :one`-style comments).
+- `schema_source` — One of:
+  - `schema_sql`: full schema as in schema.sql (for schema-based parsing).
+  - `connection_params`: DSN and options for database-only mode.
+
+**Response**
+
+Return `statements`: one `Statement` per query block. Each `Statement` has:
+
+- `name` — Query name (from `-- name: GetUser` etc.).
+- `cmd` — Command/type: use the `Cmd` enum (`engine.Cmd_CMD_ONE`, `engine.Cmd_CMD_MANY`, `engine.Cmd_CMD_EXEC`, etc.). See `protos/engine/engine.proto` for the full list.
+- `sql` — Processed SQL for that block (as-is or with `*` expanded using schema).
+- `parameters` — Parameters for this statement.
+- `columns` — Result columns (names, types, nullability, etc.) for this statement.
+
+You may also return **`catalog`** (optional). When present, sqlc passes it to the codegen plugin so codegen can emit model structs from the schema (tables/columns), not only per-query row types. Build an `engine.Catalog` from your schema (e.g. from `schema_sql` or DB metadata): one or more **CatalogSchema**, each with **CatalogTable** (rel = Identifier, columns = **CatalogColumn**). See `protos/engine/engine.proto` for the `Catalog`, `CatalogSchema`, `CatalogTable`, `CatalogColumn`, and `Identifier` messages.
+
+The engine package provides helpers (optional) to split `query.sql` and parse `"-- name: X :cmd"` lines in the same way as the built-in engines:
+
+- `engine.CommentSyntax` — Which comment styles to accept (`Dash`, `SlashStar`, `Hash`).
+- `engine.ParseNameAndCmd(line, syntax)` — Parses a single line like `"-- name: ListAuthors :many"` → `(name, cmd, ok)`. `cmd` is `engine.Cmd`.
+- `engine.QueryBlocks(content, syntax)` — Splits file content into `[]engine.QueryBlock` (each has `Name`, `Cmd`, `SQL`).
+- `engine.StatementMeta(name, cmd, sql)` — Builds a `*engine.Statement` with name/cmd/sql set; you add parameters and columns.
+
+Example handler using helpers:
+
+```go
+func handleParse(req *engine.ParseRequest) (*engine.ParseResponse, error) {
+    queryFileContent := req.GetSql()
+    syntax := engine.CommentSyntax{Dash: true, SlashStar: true, Hash: true}
+
+    var schema *SchemaInfo
+    if s := req.GetSchemaSql(); s != "" {
+        schema = parseSchema(s)
+    }
+    // Or use req.GetConnectionParams() for database-only mode.
+
+    blocks, _ := engine.QueryBlocks(queryFileContent, syntax)
+    var statements []*engine.Statement
+    for _, b := range blocks {
+        st := engine.StatementMeta(b.Name, b.Cmd, processSQL(b.SQL, schema))
+        st.Parameters = extractParameters(b.SQL)
+        st.Columns = extractColumns(b.SQL, schema)
+        statements = append(statements, st)
+    }
+    resp := &engine.ParseResponse{Statements: statements}
+    if cat := buildCatalogFromSchema(schema); cat != nil {
+        resp.Catalog = cat
+    }
+    return resp, nil
+}
+```
+
+Parameter and column types use the `Parameter` and `Column` messages in `engine.proto` (name, position, data_type, nullable, is_array, array_dims; for columns, table_name and schema_name are optional). If you return a **Catalog** (tables/columns from schema or DB), sqlc forwards it to the codegen plugin so generated code can include model types from the schema.
+
+Support for sqlc placeholders (`sqlc.arg()`, `sqlc.narg()`, `sqlc.slice()`, `sqlc.embed()`) is up to the plugin: it can parse and map them into `parameters` (and schema usage) as needed.
+
+#### Catalog (optional)
+
+If the plugin returns a non-empty **`catalog`** in `ParseResponse`, sqlc converts it to the codegen plugin’s Catalog and passes it in the GenerateRequest. Codegen plugins can then emit model structs from the schema (e.g. `type Author struct`) in addition to per-query row types. Build the catalog from `schema_sql` or from live DB metadata when using `connection_params`. The `engine.proto` defines `Catalog`, `CatalogSchema`, `CatalogTable`, `CatalogColumn`, and `Identifier` for this purpose.
+
+### 3. Build and run
+
+```bash
+go build -o sqlc-engine-external-db .
+# Ensure sqlc-engine-external-db is on PATH or use an absolute path in process.cmd
+```
+
+## Protocol
+
+Process plugins use Protocol Buffers on stdin/stdout:
+
+```
+sqlc  →  stdin (protobuf)  →  plugin  →  stdout (protobuf)  →  sqlc
+```
+
+Invocation:
+
+```bash
+sqlc-engine-external-db parse   # stdin: ParseRequest, stdout: ParseResponse
+```
+
+The definition lives in `protos/engine/engine.proto` (generated Go in `pkg/engine`). After editing the proto, run `make proto-engine-plugin` to regenerate the Go code.
+
+## Example
+
+The protocol and Go SDK are in this repository: `protos/engine/engine.proto` and `pkg/engine/` (including `sdk.go` with `engine.Run` and `engine.Handler`). Use them to build a binary that implements the Parse RPC; register it under `engines` in sqlc.yaml as shown above.
+
+## Architecture
+
+For each `sql[]` block, `sqlc generate` branches on the configured engine: built-in (postgresql, mysql, sqlite) use the compiler and catalog; any engine listed under `engines:` in sqlc.yaml uses the plugin path (no compiler). For the plugin path, sqlc calls Parse **once per query file**, sending the full file contents and schema (or connection params). The plugin returns **N statements** (one per query block) and optionally a **catalog** (tables/columns from schema or DB). sqlc passes each statement to codegen as a separate query and, when present, passes the catalog so codegen can emit model types from the schema.
+
+```
+┌─────────────────────────────────────────────────────────────────┐
+│  sqlc generate (plugin engine)                                  │
+│  1. Per query file: one Parse(schema_sql|connection_params,      │
+│     full query file content)                                    │
+│  2. ParseResponse.statements = one Statement per query block     │
+│  3. Each statement → one codegen query (N helpers)               │
+└─────────────────────────────────────────────────────────────────┘
+
+    sqlc                          sqlc-engine-external-db
+      │──── spawn, args: ["parse"] ──────────────────────────────► │
+      │──── stdin: ParseRequest{sql=full query.sql, schema_sql|…}  ► │
+      │◄─── stdout: ParseResponse{statements: [stmt1, stmt2, …]} ── │
+```
+
+## See also
+
+- [Codegen plugins](plugins.md) — Custom code generators that consume engine output.
+- [Configuration reference](../reference/config.md)
+- Proto schema: `protos/engine/engine.proto`
diff --git a/docs/index.rst b/docs/index.rst
index f914f3ec41..927c5de1f2 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -96,5 +96,6 @@ code ever again.
    howto/ci-cd.md
    guides/using-go-and-pgx.rst
    guides/plugins.md
+   guides/engine-plugins.md
    guides/development.md
    guides/privacy.md
diff --git a/internal/cmd/generate.go b/internal/cmd/generate.go
index 05b5445ebb..18f1b74bf4 100644
--- a/internal/cmd/generate.go
+++ b/internal/cmd/generate.go
@@ -131,6 +131,17 @@ func readConfig(stderr io.Writer, dir, filename string) (string, *config.Config,
 	return configPath, &conf, nil
 }
 
+// sourceFiles holds in-memory config and optional file contents so generate can run
+// without reading from disk (e.g. in tests). For production, Generate reads from FS and
+// fills FileContents before calling generate.
+type sourceFiles struct {
+	Config       *config.Config
+	ConfigPath   string
+	Dir          string
+	FileContents map[string][]byte // path -> content; keys match paths used when reading (e.g. filepath.Join(dir, "schema.sql"))
+}
+
+// Generate runs codegen for the given directory and config file, reading all input from disk.
 func Generate(ctx context.Context, dir, filename string, o *Options) (map[string]string, error) {
 	e := o.Env
 	stderr := o.Stderr
@@ -151,46 +162,95 @@ func Generate(ctx context.Context, dir, filename string, o *Options) (map[string
 		return nil, err
 	}
 
-	// Comment on why these two methods exist
 	if conf.Cloud.Project != "" && e.Remote && !e.NoRemote {
 		return remoteGenerate(ctx, configPath, conf, dir, stderr)
 	}
 
+	inputs := &sourceFiles{Config: conf, ConfigPath: configPath, Dir: dir}
+	inputs.FileContents, err = loadFileContentsFromFS(conf, dir)
+	if err != nil {
+		return nil, err
+	}
+	return generate(ctx, inputs, o)
+}
+
+// generate performs codegen using in-memory inputs. It is used by Generate (with contents
+// loaded from disk) and by tests (with pre-filled Config and FileContents, no temp files).
+func generate(ctx context.Context, inputs *sourceFiles, o *Options) (map[string]string, error) {
 	g := &generator{
-		dir:    dir,
+		dir:    inputs.Dir,
 		output: map[string]string{},
 	}
-
-	if err := processQuerySets(ctx, g, conf, dir, o); err != nil {
+	if o != nil && o.CodegenHandlerOverride != nil {
+		g.codegenHandlerOverride = o.CodegenHandlerOverride
+	}
+	if err := processQuerySets(ctx, g, inputs, o); err != nil {
 		return nil, err
 	}
-
 	return g.output, nil
 }
 
+// loadFileContentsFromFS reads all schema and query files referenced in conf into a map
+// path -> content, using dir to resolve paths.
+func loadFileContentsFromFS(conf *config.Config, dir string) (map[string][]byte, error) {
+	out := make(map[string][]byte)
+	for _, pkg := range conf.SQL {
+		for _, rel := range pkg.Schema {
+			path := filepath.Join(dir, rel)
+			files, err := sqlpath.Glob([]string{path})
+			if err != nil {
+				return nil, err
+			}
+			for _, f := range files {
+				b, err := os.ReadFile(f)
+				if err != nil {
+					return nil, err
+				}
+				out[f] = b
+			}
+		}
+		for _, rel := range pkg.Queries {
+			path := filepath.Join(dir, rel)
+			files, err := sqlpath.Glob([]string{path})
+			if err != nil {
+				return nil, err
+			}
+			for _, f := range files {
+				b, err := os.ReadFile(f)
+				if err != nil {
+					return nil, err
+				}
+				out[f] = b
+			}
+		}
+	}
+	return out, nil
+}
+
 type generator struct {
-	m      sync.Mutex
-	dir    string
-	output map[string]string
+	m                      sync.Mutex
+	dir                    string
+	output                 map[string]string
+	codegenHandlerOverride grpc.ClientConnInterface
 }
 
-func (g *generator) Pairs(ctx context.Context, conf *config.Config) []OutputPair {
-	var pairs []OutputPair
+func (g *generator) Pairs(ctx context.Context, conf *config.Config) []outputPair {
+	var pairs []outputPair
 	for _, sql := range conf.SQL {
 		if sql.Gen.Go != nil {
-			pairs = append(pairs, OutputPair{
+			pairs = append(pairs, outputPair{
 				SQL: sql,
 				Gen: config.SQLGen{Go: sql.Gen.Go},
 			})
 		}
 		if sql.Gen.JSON != nil {
-			pairs = append(pairs, OutputPair{
+			pairs = append(pairs, outputPair{
 				SQL: sql,
 				Gen: config.SQLGen{JSON: sql.Gen.JSON},
 			})
 		}
 		for i := range sql.Codegen {
-			pairs = append(pairs, OutputPair{
+			pairs = append(pairs, outputPair{
 				SQL:    sql,
 				Plugin: &sql.Codegen[i],
 			})
@@ -199,8 +259,8 @@ func (g *generator) Pairs(ctx context.Context, conf *config.Config) []OutputPair
 	return pairs
 }
 
-func (g *generator) ProcessResult(ctx context.Context, combo config.CombinedSettings, sql OutputPair, result *compiler.Result) error {
-	out, resp, err := codegen(ctx, combo, sql, result)
+func (g *generator) ProcessResult(ctx context.Context, combo config.CombinedSettings, sql outputPair, result *compiler.Result) error {
+	out, resp, err := codegen(ctx, combo, sql, result, g.codegenHandlerOverride)
 	if err != nil {
 		return err
 	}
@@ -333,7 +393,7 @@ func parse(ctx context.Context, name, dir string, sql config.SQL, combo config.C
 	return c.Result(), false
 }
 
-func codegen(ctx context.Context, combo config.CombinedSettings, sql OutputPair, result *compiler.Result) (string, *plugin.GenerateResponse, error) {
+func codegen(ctx context.Context, combo config.CombinedSettings, sql outputPair, result *compiler.Result, codegenOverride grpc.ClientConnInterface) (string, *plugin.GenerateResponse, error) {
 	defer trace.StartRegion(ctx, "codegen").End()
 	req := codeGenRequest(result, combo)
 	var handler grpc.ClientConnInterface
@@ -341,42 +401,47 @@ func codegen(ctx context.Context, combo config.CombinedSettings, sql OutputPair,
 	switch {
 	case sql.Plugin != nil:
 		out = sql.Plugin.Out
-		plug, err := findPlugin(combo.Global, sql.Plugin.Plugin)
-		if err != nil {
-			return "", nil, fmt.Errorf("plugin not found: %s", err)
-		}
+		if codegenOverride != nil {
+			handler = codegenOverride
+		} else {
+			plug, err := findPlugin(combo.Global, sql.Plugin.Plugin)
+			if err != nil {
+				return "", nil, fmt.Errorf("plugin not found: %s", err)
+			}
 
-		switch {
-		case plug.Process != nil:
-			handler = &process.Runner{
-				Cmd:    plug.Process.Cmd,
-				Env:    plug.Env,
-				Format: plug.Process.Format,
+			switch {
+			case plug.Process != nil:
+				handler = &process.Runner{
+					Cmd:    plug.Process.Cmd,
+					Env:    plug.Env,
+					Format: plug.Process.Format,
+				}
+			case plug.WASM != nil:
+				handler = &wasm.Runner{
+					URL:    plug.WASM.URL,
+					SHA256: plug.WASM.SHA256,
+					Env:    plug.Env,
+				}
+			default:
+				return "", nil, fmt.Errorf("unsupported plugin type")
 			}
-		case plug.WASM != nil:
-			handler = &wasm.Runner{
-				URL:    plug.WASM.URL,
-				SHA256: plug.WASM.SHA256,
-				Env:    plug.Env,
+			global, found := combo.Global.Options[plug.Name]
+			if found {
+				opts, err := convert.YAMLtoJSON(global)
+				if err != nil {
+					return "", nil, fmt.Errorf("invalid global options: %w", err)
+				}
+				req.GlobalOptions = opts
 			}
-		default:
-			return "", nil, fmt.Errorf("unsupported plugin type")
 		}
-
 		opts, err := convert.YAMLtoJSON(sql.Plugin.Options)
 		if err != nil {
 			return "", nil, fmt.Errorf("invalid plugin options: %w", err)
 		}
-		req.PluginOptions = opts
-
-		global, found := combo.Global.Options[plug.Name]
-		if found {
-			opts, err := convert.YAMLtoJSON(global)
-			if err != nil {
-				return "", nil, fmt.Errorf("invalid global options: %w", err)
-			}
-			req.GlobalOptions = opts
+		if err := validateExternalPluginOptions(opts, sql.Plugin.Plugin); err != nil {
+			return "", nil, err
 		}
+		req.PluginOptions = opts
 
 	case sql.Gen.Go != nil:
 		out = combo.Go.Out
@@ -411,3 +476,28 @@ func codegen(ctx context.Context, combo config.CombinedSettings, sql OutputPair,
 	resp, err := client.Generate(ctx, req)
 	return out, resp, err
 }
+
+// driverOnlyOptions are options that apply only to built-in Go codegen, not to external plugins.
+var driverOnlyOptions = []string{"sql_package", "sql_driver"}
+
+// validateExternalPluginOptions returns an error if plugin options contain sql_package or sql_driver.
+// External codegen plugins define their own database driver; these options are not supported.
+func validateExternalPluginOptions(opts []byte, pluginName string) error {
+	if len(opts) == 0 {
+		return nil
+	}
+	var m map[string]interface{}
+	if err := json.Unmarshal(opts, &m); err != nil {
+		return nil
+	}
+	var invalid []string
+	for _, key := range driverOnlyOptions {
+		if _, ok := m[key]; ok {
+			invalid = append(invalid, key)
+		}
+	}
+	if len(invalid) == 0 {
+		return nil
+	}
+	return fmt.Errorf("plugin %q: options %q are not supported for external codegen plugins; the plugin defines its own database driver (these options only apply to built-in Go codegen)", pluginName, invalid)
+}
diff --git a/internal/cmd/options.go b/internal/cmd/options.go
index 02d3614f4e..78ccb2f818 100644
--- a/internal/cmd/options.go
+++ b/internal/cmd/options.go
@@ -3,7 +3,10 @@ package cmd
 import (
 	"io"
 
+	"google.golang.org/grpc"
+
 	"github.com/sqlc-dev/sqlc/internal/config"
+	pb "github.com/sqlc-dev/sqlc/pkg/engine"
 )
 
 type Options struct {
@@ -15,6 +18,10 @@ type Options struct {
 
 	// Testing only
 	MutateConfig func(*config.Config)
+	// CodegenHandlerOverride injects a mock codegen handler instead of spawning a process.
+	CodegenHandlerOverride grpc.ClientConnInterface
+	// PluginParseFunc, when set, is used in the plugin-engine path instead of invoking the engine process (for tests).
+	PluginParseFunc func(schemaSQL, querySQL string) (*pb.ParseResponse, error)
 }
 
 func (o *Options) ReadConfig(dir, filename string) (string, *config.Config, error) {
diff --git a/internal/cmd/plugin_engine.go b/internal/cmd/plugin_engine.go
new file mode 100644
index 0000000000..e23b6c4e69
--- /dev/null
+++ b/internal/cmd/plugin_engine.go
@@ -0,0 +1,339 @@
+// This file runs a database-engine plugin as an external process (parse RPC over stdin/stdout).
+// It is used only by the plugin-engine generate path (runPluginQuerySet). Vet does not support plugin engines.
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+
+	"github.com/sqlc-dev/sqlc/internal/compiler"
+	"github.com/sqlc-dev/sqlc/internal/config"
+	"github.com/sqlc-dev/sqlc/internal/metadata"
+	"github.com/sqlc-dev/sqlc/internal/multierr"
+	"github.com/sqlc-dev/sqlc/internal/plugin"
+	"github.com/sqlc-dev/sqlc/internal/sql/ast"
+	"github.com/sqlc-dev/sqlc/internal/sql/catalog"
+	"github.com/sqlc-dev/sqlc/internal/sql/sqlpath"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/sqlc-dev/sqlc/internal/info"
+	pb "github.com/sqlc-dev/sqlc/pkg/engine"
+)
+
+// engineProcessRunner runs an engine plugin as an external process.
+type engineProcessRunner struct {
+	Cmd string
+	Dir string // Working directory for the plugin (config file directory)
+	Env []string
+}
+
+func newEngineProcessRunner(cmd, dir string, env []string) *engineProcessRunner {
+	return &engineProcessRunner{Cmd: cmd, Dir: dir, Env: env}
+}
+
+func (r *engineProcessRunner) invoke(ctx context.Context, method string, req, resp proto.Message) error {
+	stdin, err := proto.Marshal(req)
+	if err != nil {
+		return fmt.Errorf("failed to encode request: %w", err)
+	}
+
+	cmdParts := strings.Fields(r.Cmd)
+	if len(cmdParts) == 0 {
+		return fmt.Errorf("engine plugin not found: %s\n\nSee the engine plugins documentation: https://docs.sqlc.dev/en/latest/guides/engine-plugins.html", r.Cmd)
+	}
+
+	path, err := exec.LookPath(cmdParts[0])
+	if err != nil {
+		return fmt.Errorf("engine plugin not found: %s\n\nSee the engine plugins documentation: https://docs.sqlc.dev/en/latest/guides/engine-plugins.html", r.Cmd)
+	}
+
+	args := append(cmdParts[1:], method)
+	cmd := exec.CommandContext(ctx, path, args...)
+	cmd.Stdin = bytes.NewReader(stdin)
+	if r.Dir != "" {
+		cmd.Dir = r.Dir
+	}
+	cmd.Env = append(os.Environ(), fmt.Sprintf("SQLC_VERSION=%s", info.Version))
+
+	out, err := cmd.Output()
+	if err != nil {
+		stderr := err.Error()
+		var exit *exec.ExitError
+		if errors.As(err, &exit) {
+			stderr = string(exit.Stderr)
+		}
+		return fmt.Errorf("engine plugin error: %s", stderr)
+	}
+
+	if err := proto.Unmarshal(out, resp); err != nil {
+		return fmt.Errorf("failed to decode response: %w", err)
+	}
+	return nil
+}
+
+// parseRequest invokes the plugin's Parse RPC. Used by runPluginQuerySet.
+func (r *engineProcessRunner) parseRequest(ctx context.Context, req *pb.ParseRequest) (*pb.ParseResponse, error) {
+	resp := &pb.ParseResponse{}
+	if err := r.invoke(ctx, "parse", req, resp); err != nil {
+		return nil, err
+	}
+	return resp, nil
+}
+
+// runPluginQuerySet runs the plugin-engine path: schema and queries are sent to the
+// engine plugin via ParseRequest; the responses are turned into compiler.Result and
+// passed to ProcessResult. No AST or compiler parsing is used.
+// When inputs.FileContents is set, schema/query bytes are taken from it (no disk read).
+func runPluginQuerySet(ctx context.Context, rp resultProcessor, name, dir string, sql outputPair, combo config.CombinedSettings, inputs *sourceFiles, o *Options) error {
+	enginePlugin, found := config.FindEnginePlugin(&combo.Global, string(combo.Package.Engine))
+	if !found || enginePlugin.Process == nil {
+		e := string(combo.Package.Engine)
+		return fmt.Errorf("unknown engine: %s\n\nAdd the engine to the 'engines' section of sqlc.yaml. See the engine plugins documentation: https://docs.sqlc.dev/en/latest/guides/engine-plugins.html", e)
+	}
+
+	readFile := func(path string) ([]byte, error) {
+		if inputs != nil && inputs.FileContents != nil {
+			if b, ok := inputs.FileContents[path]; ok {
+				return b, nil
+			}
+		}
+		return os.ReadFile(path)
+	}
+
+	databaseOnly := combo.Package.Analyzer.Database.IsOnly() && combo.Package.Database != nil && combo.Package.Database.URI != ""
+
+	var schemaSQL string
+	if !databaseOnly {
+		var err error
+		if inputs != nil && inputs.FileContents != nil {
+			var parts []string
+			for _, p := range sql.Schema {
+				if b, ok := inputs.FileContents[p]; ok {
+					parts = append(parts, string(b))
+				}
+			}
+			schemaSQL = strings.Join(parts, "\n")
+		} else {
+			schemaSQL, err = loadSchemaSQL(sql.Schema, readFile)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	type parseFuncType func(querySQL string) (*pb.ParseResponse, error)
+	var parseFn parseFuncType
+	if o != nil && o.PluginParseFunc != nil {
+		schemaStr := schemaSQL
+		if databaseOnly {
+			schemaStr = ""
+		}
+		parseFn = func(querySQL string) (*pb.ParseResponse, error) {
+			return o.PluginParseFunc(schemaStr, querySQL)
+		}
+	} else {
+		r := newEngineProcessRunner(enginePlugin.Process.Cmd, combo.Dir, enginePlugin.Env)
+		parseFn = func(querySQL string) (*pb.ParseResponse, error) {
+			req := &pb.ParseRequest{Sql: querySQL}
+			if databaseOnly {
+				req.SchemaSource = &pb.ParseRequest_ConnectionParams{
+					ConnectionParams: &pb.ConnectionParams{Dsn: combo.Package.Database.URI},
+				}
+			} else {
+				req.SchemaSource = &pb.ParseRequest_SchemaSql{SchemaSql: schemaSQL}
+			}
+			return r.parseRequest(ctx, req)
+		}
+	}
+
+	var queryPaths []string
+	var err error
+	if inputs != nil && inputs.FileContents != nil {
+		queryPaths = sql.Queries
+	} else {
+		queryPaths, err = sqlpath.Glob(sql.Queries)
+		if err != nil {
+			return err
+		}
+	}
+
+	var queries []*compiler.Query
+	var pluginCatalog *plugin.Catalog
+	merr := multierr.New()
+	set := map[string]struct{}{}
+
+	for _, filename := range queryPaths {
+		blob, err := readFile(filename)
+		if err != nil {
+			merr.Add(filename, "", 0, err)
+			continue
+		}
+		queryContent := string(blob)
+		resp, err := parseFn(queryContent)
+		if err != nil {
+			merr.Add(filename, queryContent, 0, err)
+			continue
+		}
+		if pluginCatalog == nil && resp.GetCatalog() != nil {
+			pluginCatalog = engineCatalogToPlugin(resp.GetCatalog())
+		}
+		baseName := filepath.Base(filename)
+		stmts := resp.GetStatements()
+		for _, st := range stmts {
+			q := statementToCompilerQuery(st, baseName)
+			if q == nil {
+				continue
+			}
+			qName := st.GetName()
+			if _, exists := set[qName]; exists {
+				merr.Add(filename, queryContent, 0, fmt.Errorf("duplicate query name: %s", qName))
+				continue
+			}
+			set[qName] = struct{}{}
+			queries = append(queries, q)
+		}
+	}
+
+	if len(merr.Errs()) > 0 {
+		return merr
+	}
+	if len(queries) == 0 {
+		return fmt.Errorf("no queries in paths %s", strings.Join(sql.Queries, ","))
+	}
+
+	result := &compiler.Result{
+		Catalog:       catalog.New(""),
+		Queries:       queries,
+		PluginCatalog: pluginCatalog,
+	}
+	return rp.ProcessResult(ctx, combo, sql, result)
+}
+
+func loadSchemaSQL(schemaPaths []string, readFile func(string) ([]byte, error)) (string, error) {
+	var parts []string
+	for _, p := range schemaPaths {
+		files, err := sqlpath.Glob([]string{p})
+		if err != nil {
+			return "", err
+		}
+		if len(files) == 0 {
+			files = []string{p}
+		}
+		for _, f := range files {
+			b, err := readFile(f)
+			if err != nil {
+				return "", err
+			}
+			parts = append(parts, string(b))
+		}
+	}
+	return strings.Join(parts, "\n"), nil
+}
+
+// engineCatalogToPlugin converts engine.Catalog (from ParseResponse) to plugin.Catalog for codegen.
+func engineCatalogToPlugin(engineCat *pb.Catalog) *plugin.Catalog {
+	if engineCat == nil {
+		return nil
+	}
+	var schemas []*plugin.Schema
+	for _, es := range engineCat.GetSchemas() {
+		var tables []*plugin.Table
+		for _, et := range es.GetTables() {
+			rel := et.GetRel()
+			if rel == nil {
+				continue
+			}
+			var cols []*plugin.Column
+			for _, col := range et.GetColumns() {
+				cols = append(cols, &plugin.Column{
+					Name:      col.GetName(),
+					NotNull:   col.GetNotNull(),
+					IsArray:   col.GetIsArray(),
+					ArrayDims: col.GetArrayDims(),
+					Type:      engineIDToPlugin(col.GetType()),
+					Table:     engineIDToPlugin(rel),
+				})
+			}
+			tables = append(tables, &plugin.Table{
+				Rel:     engineIDToPlugin(rel),
+				Columns: cols,
+				Comment: et.GetComment(),
+			})
+		}
+		schemas = append(schemas, &plugin.Schema{
+			Name:    es.GetName(),
+			Tables:  tables,
+			Comment: es.GetComment(),
+		})
+	}
+	return &plugin.Catalog{
+		Comment:       engineCat.GetComment(),
+		DefaultSchema: engineCat.GetDefaultSchema(),
+		Name:          engineCat.GetName(),
+		Schemas:       schemas,
+	}
+}
+
+func engineIDToPlugin(e *pb.Identifier) *plugin.Identifier {
+	if e == nil {
+		return nil
+	}
+	return &plugin.Identifier{Catalog: e.GetCatalog(), Schema: e.GetSchema(), Name: e.GetName()}
+}
+
+// statementToCompilerQuery converts one engine.Statement from the plugin into a compiler.Query.
+func statementToCompilerQuery(st *pb.Statement, filename string) *compiler.Query {
+	if st == nil {
+		return nil
+	}
+	sqlTrimmed := strings.TrimSpace(st.GetSql())
+	if sqlTrimmed == "" {
+		return nil
+	}
+	var params []compiler.Parameter
+	for _, p := range st.GetParameters() {
+		col := &compiler.Column{
+			Name:      p.GetName(),
+			DataType:  p.GetDataType(),
+			NotNull:   !p.GetNullable(),
+			IsArray:   p.GetIsArray(),
+			ArrayDims: int(p.GetArrayDims()),
+		}
+		pos := int(p.GetPosition())
+		if pos < 1 {
+			pos = len(params) + 1
+		}
+		params = append(params, compiler.Parameter{Number: pos, Column: col})
+	}
+	var columns []*compiler.Column
+	for _, c := range st.GetColumns() {
+		col := &compiler.Column{
+			Name:      c.GetName(),
+			DataType:  c.GetDataType(),
+			NotNull:   !c.GetNullable(),
+			IsArray:   c.GetIsArray(),
+			ArrayDims: int(c.GetArrayDims()),
+		}
+		if tn := c.GetTableName(); tn != "" {
+			col.Table = &ast.TableName{Name: tn}
+		}
+		columns = append(columns, col)
+	}
+	return &compiler.Query{
+		SQL: sqlTrimmed,
+		Metadata: metadata.Metadata{
+			Name:     st.GetName(),
+			Cmd:      pb.CmdToString(st.GetCmd()),
+			Filename: filename,
+		},
+		Params:  params,
+		Columns: columns,
+	}
+}
diff --git a/internal/cmd/plugin_engine_test.go b/internal/cmd/plugin_engine_test.go
new file mode 100644
index 0000000000..73583ab2d1
--- /dev/null
+++ b/internal/cmd/plugin_engine_test.go
@@ -0,0 +1,413 @@
+package cmd
+
+// Engine-plugin pipeline integration tests.
+//
+// Why here (cmd) and not in endtoend?
+//   - endtoend: black-box replay of full sqlc on testdata dirs, comparing stdout/stderr/output to golden files.
+//   - These tests: unit-style integration of the engine-plugin path inside cmd — in-memory config and FileContents,
+//     mocks for engine and codegen, no temp dirs or real plugins. They assert the data flow (schema+query → engine
+//     mock → codegen request) and that the plugin package is used when PluginParseFunc is nil.
+//
+// Proof that the technology works:
+//   - TestPluginPipeline_FullPipeline: one block → one Parse call; that call receives schema; codegen gets the result.
+//   - TestPluginPipeline_NBlocksNCalls: N blocks in query.sql → exactly N Parse calls; each call receives schema.
+//   - TestPluginPipeline_DatabaseOnly_ReceivesNoSchema: with analyzer.database: only + database.uri, each Parse
+//     call receives empty schema (the real runner would get connection_params in ParseRequest).
+//   - TestPluginPipeline_WithoutOverride_UsesPluginPackage: with PluginParseFunc nil, generate fails with an error
+//     that is NOT "unknown engine", so we did enter runPluginQuerySet and call the engine process runner.
+
+import (
+	"bytes"
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/sqlc-dev/sqlc/internal/config"
+	"github.com/sqlc-dev/sqlc/internal/ext"
+	"github.com/sqlc-dev/sqlc/internal/opts"
+	"github.com/sqlc-dev/sqlc/internal/plugin"
+	pb "github.com/sqlc-dev/sqlc/pkg/engine"
+)
+
+const testPluginPipelineConfig = `version: "2"
+sql:
+  - engine: "testeng"
+    schema: ["schema.sql"]
+    queries: ["query.sql"]
+    codegen:
+      - plugin: "mock"
+        out: "."
+plugins:
+  - name: "mock"
+    process:
+      cmd: "echo"
+engines:
+  - name: "testeng"
+    process:
+      cmd: "echo"
+`
+
+// engineMockRecord holds what the engine-plugin mock received and returned.
+// Used to validate that the pipeline passes schema + raw query in, and that
+// the plugin's Sql/Parameters/Columns are what later reach codegen.
+// CalledWith records each Parse call so we can assert N blocks → N calls and
+// that every call received schema (or "" in databaseOnly mode).
+type engineMockRecord struct {
+	Calls          int
+	SchemaSQL      string // last call (backward compat)
+	QuerySQL       string // last call (backward compat)
+	CalledWith     []struct{ SchemaSQL, QuerySQL string }
+	ReturnedSQL    string
+	ReturnedParams []*pb.Parameter
+	ReturnedCols   []*pb.Column
+}
+
+// codegenMockRecord holds what the codegen-plugin mock received.
+type codegenMockRecord struct {
+	Request *plugin.GenerateRequest
+}
+
+// TestPluginPipeline_FullPipeline validates the plugin-engine data flow end to end:
+//
+//  1. Inputs: schema and query file contents (from FileContents) are passed into the
+//     engine-plugin mock. We assert exactly what the mock received.
+//  2. Engine plugin returns: Sql, Parameters, Columns (the "enriched" result).
+//  3. Pipeline converts that into compiler.Result and then into plugin.GenerateRequest.
+//  4. Codegen plugin receives that request. We assert it contains the same SQL, params,
+//     and columns, and that query name/cmd come from the query file comments.
+//
+// Note: the engine process runner is not called here because we use PluginParseFunc.
+// That mock replaces the real ProcessRunner entirely. This test validates the cmd
+// pipeline and the data contract at the boundaries; coverage of the plugin package
+// comes from other tests (e.g. process runner, or an E2E test with a real engine binary).
+func TestPluginPipeline_FullPipeline(t *testing.T) {
+	ctx := context.Background()
+
+	// --- Inputs (what we feed into the pipeline via FileContents) ---
+	const (
+		schemaContent = "CREATE TABLE users (id INT, name TEXT);"
+		queryContent  = "-- name: GetUser :one\nSELECT id, name FROM users WHERE id = $1"
+	)
+
+	// --- Engine mock: record inputs, return "enriched" output ---
+	engineRecord := &engineMockRecord{
+		ReturnedSQL: "SELECT id, name FROM users WHERE id = $1",
+		ReturnedParams: []*pb.Parameter{
+			{Position: 1, DataType: "int", Nullable: false},
+		},
+		ReturnedCols: []*pb.Column{
+			{Name: "id", DataType: "int", Nullable: false},
+			{Name: "name", DataType: "text", Nullable: false},
+		},
+	}
+	pluginParse := func(schemaSQL, querySQL string) (*pb.ParseResponse, error) {
+		engineRecord.Calls++
+		engineRecord.SchemaSQL = schemaSQL
+		engineRecord.QuerySQL = querySQL
+		engineRecord.CalledWith = append(engineRecord.CalledWith, struct{ SchemaSQL, QuerySQL string }{schemaSQL, querySQL})
+		return &pb.ParseResponse{
+			Statements: []*pb.Statement{{
+				Name: "GetUser", Cmd: pb.Cmd_CMD_ONE, Sql: engineRecord.ReturnedSQL,
+				Parameters: engineRecord.ReturnedParams, Columns: engineRecord.ReturnedCols,
+			}},
+		}, nil
+	}
+
+	// --- Codegen mock: record the full request ---
+	codegenRecord := &codegenMockRecord{}
+	mockCodegen := ext.HandleFunc(func(_ context.Context, req *plugin.GenerateRequest) (*plugin.GenerateResponse, error) {
+		codegenRecord.Request = req
+		return &plugin.GenerateResponse{}, nil
+	})
+
+	conf, err := config.ParseConfig(strings.NewReader(testPluginPipelineConfig))
+	if err != nil {
+		t.Fatalf("parse config: %v", err)
+	}
+
+	inputs := &sourceFiles{
+		Config:     &conf,
+		ConfigPath: "sqlc.yaml",
+		Dir:        ".",
+		FileContents: map[string][]byte{
+			"schema.sql": []byte(schemaContent),
+			"query.sql":  []byte(queryContent),
+		},
+	}
+
+	var stderr bytes.Buffer
+	debug := opts.DebugFromString("")
+	debug.ProcessPlugins = true
+	o := &Options{
+		Env:                    Env{Debug: debug},
+		Stderr:                 &stderr,
+		PluginParseFunc:        pluginParse,
+		CodegenHandlerOverride: mockCodegen,
+	}
+
+	_, err = generate(ctx, inputs, o)
+	if err != nil {
+		t.Fatalf("generate failed: %v\nstderr: %s", err, stderr.String())
+	}
+
+	// ---- 1) Validate what was sent INTO the engine plugin ----
+	// N blocks in query.sql must yield N Parse calls; each call must receive schema (or connection in databaseOnly).
+	if engineRecord.Calls != 1 {
+		t.Errorf("engine mock called %d times, want 1 (one block → one Parse call)", engineRecord.Calls)
+	}
+	if len(engineRecord.CalledWith) != 1 {
+		t.Errorf("engine mock CalledWith has %d entries, want 1", len(engineRecord.CalledWith))
+	}
+	if len(engineRecord.CalledWith) > 0 && engineRecord.CalledWith[0].SchemaSQL != schemaContent {
+		t.Errorf("every Parse call must receive schema: got %q", engineRecord.CalledWith[0].SchemaSQL)
+	}
+	if engineRecord.SchemaSQL != schemaContent {
+		t.Errorf("engine received schema:\n  got:  %q\n  want: %q", engineRecord.SchemaSQL, schemaContent)
+	}
+	// With one block, query SQL is the whole block (same as queryContent)
+	if engineRecord.QuerySQL != queryContent {
+		t.Errorf("engine received query:\n  got:  %q\n  want: %q", engineRecord.QuerySQL, queryContent)
+	}
+
+	// ---- 2) Validate what the codegen plugin received (must match engine output + metadata) ----
+	if codegenRecord.Request == nil {
+		t.Fatal("codegen mock was never called; request not recorded")
+	}
+	if len(codegenRecord.Request.Queries) == 0 {
+		t.Fatal("codegen request has no queries")
+	}
+	q := codegenRecord.Request.Queries[0]
+
+	// Name and Cmd come from the query file comment "-- name: GetUser :one"
+	if got := q.GetName(); got != "GetUser" {
+		t.Errorf("codegen query name = %q, want %q", got, "GetUser")
+	}
+	if got := q.GetCmd(); got != ":one" {
+		t.Errorf("codegen query cmd = %q, want %q", got, ":one")
+	}
+
+	// Text must be exactly what the engine plugin returned (pipeline does not change it)
+	if q.GetText() == "" {
+		t.Error("codegen query has empty Text; plugin Sql did not reach codegen")
+	}
+	if got := q.GetText(); got != engineRecord.ReturnedSQL {
+		t.Errorf("codegen query Text = %q, want (engine output) %q", got, engineRecord.ReturnedSQL)
+	}
+
+	// Params and columns must match what the engine plugin returned (codegen receives unchanged)
+	if len(q.GetParams()) != len(engineRecord.ReturnedParams) {
+		t.Errorf("codegen query has %d params, want %d", len(q.GetParams()), len(engineRecord.ReturnedParams))
+	} else {
+		for i, want := range engineRecord.ReturnedParams {
+			p := q.GetParams()[i]
+			if p.GetNumber() != want.Position {
+				t.Errorf("param[%d] number = %d, want %d", i, p.GetNumber(), want.Position)
+			}
+			// plugin.Parameter.Column.Type.Name holds the data type
+			if col := p.GetColumn(); col != nil && col.GetType() != nil {
+				if got := col.GetType().GetName(); got != want.DataType {
+					t.Errorf("param[%d] DataType = %q, want %q", i, got, want.DataType)
+				}
+			}
+		}
+	}
+	if len(q.GetColumns()) != len(engineRecord.ReturnedCols) {
+		t.Errorf("codegen query has %d columns, want %d", len(q.GetColumns()), len(engineRecord.ReturnedCols))
+	} else {
+		for i, want := range engineRecord.ReturnedCols {
+			c := q.GetColumns()[i]
+			if c.GetName() != want.Name {
+				t.Errorf("column[%d] name = %q, want %q", i, c.GetName(), want.Name)
+			}
+			if typ := c.GetType(); typ != nil && typ.GetName() != want.DataType {
+				t.Errorf("column[%d] type = %q, want %q", i, typ.GetName(), want.DataType)
+			}
+		}
+	}
+
+	// Sanity: codegen received exactly one query and we validated it
+	if len(codegenRecord.Request.Queries) != 1 {
+		t.Errorf("codegen received %d queries, expected 1", len(codegenRecord.Request.Queries))
+	}
+}
+
+// TestPluginPipeline_WithoutOverride_UsesPluginPackage proves that when PluginParseFunc
+// is not set, the pipeline calls the engine process runner (newEngineProcessRunner + parseRequest).
+// It runs generate with a plugin engine and nil PluginParseFunc; we expect failure
+// (e.g. from running "echo" as the engine binary), but the error must NOT be
+// "unknown engine" — so we know we went past config lookup and into the plugin path.
+// If you add panic("azaza") at the start of newEngineProcessRunner or parseRequest,
+// this test will panic, confirming that the plugin package is actually invoked.
+func TestPluginPipeline_WithoutOverride_UsesPluginPackage(t *testing.T) {
+	ctx := context.Background()
+	conf, err := config.ParseConfig(strings.NewReader(testPluginPipelineConfig))
+	if err != nil {
+		t.Fatalf("parse config: %v", err)
+	}
+	inputs := &sourceFiles{
+		Config:     &conf,
+		ConfigPath: "sqlc.yaml",
+		Dir:        ".",
+		FileContents: map[string][]byte{
+			"schema.sql": []byte("CREATE TABLE t (id INT);"),
+			"query.sql":  []byte("-- name: Get :one\nSELECT 1"),
+		},
+	}
+	var stderr bytes.Buffer
+	debug := opts.DebugFromString("")
+	debug.ProcessPlugins = true
+	o := &Options{
+		Env:             Env{Debug: debug},
+		Stderr:          &stderr,
+		PluginParseFunc: nil, // do not override — must use built-in engine process runner
+		CodegenHandlerOverride: ext.HandleFunc(func(_ context.Context, _ *plugin.GenerateRequest) (*plugin.GenerateResponse, error) {
+			return &plugin.GenerateResponse{}, nil
+		}),
+	}
+
+	_, err = generate(ctx, inputs, o)
+
+	// We expect some error (e.g. "echo" does not speak the engine protocol).
+	// What we must NOT see is "unknown engine" — that would mean we never reached
+	// the plugin path. So the plugin package was used (ParseRequest or NewProcessRunner ran).
+	if err == nil {
+		t.Fatal("expected generate to fail when using real plugin runner with cmd=echo; nil error means plugin path was not exercised as intended")
+	}
+	if strings.Contains(err.Error(), "unknown engine") {
+		t.Errorf("error is %q — we never entered the plugin path. With PluginParseFunc=nil, runPluginQuerySet must call the engine process runner.", err.Error())
+	}
+}
+
+// TestPluginPipeline_NBlocksNCalls verifies that one Parse call receives full query.sql and schema,
+// and the plugin returns N statements (one per query block); sqlc then generates N helpers.
+func TestPluginPipeline_NBlocksNCalls(t *testing.T) {
+	const (
+		schemaContent = "CREATE TABLE users (id INT, name TEXT);"
+		block1        = "-- name: GetUser :one\nSELECT id, name FROM users WHERE id = $1"
+		block2        = "-- name: ListUsers :many\nSELECT id, name FROM users ORDER BY id"
+	)
+	queryContent := block1 + "\n\n" + block2
+
+	engineRecord := &engineMockRecord{
+		ReturnedParams: []*pb.Parameter{{Position: 1, DataType: "int", Nullable: false}},
+		ReturnedCols:   []*pb.Column{{Name: "id", DataType: "int", Nullable: false}, {Name: "name", DataType: "text", Nullable: false}},
+	}
+	var codegenReq *plugin.GenerateRequest
+	pluginParse := func(schemaSQL, querySQL string) (*pb.ParseResponse, error) {
+		engineRecord.Calls++
+		engineRecord.CalledWith = append(engineRecord.CalledWith, struct{ SchemaSQL, QuerySQL string }{schemaSQL, querySQL})
+		return &pb.ParseResponse{
+			Statements: []*pb.Statement{
+				{Name: "GetUser", Cmd: pb.Cmd_CMD_ONE, Sql: "SELECT id, name FROM users WHERE id = $1", Parameters: engineRecord.ReturnedParams, Columns: engineRecord.ReturnedCols},
+				{Name: "ListUsers", Cmd: pb.Cmd_CMD_MANY, Sql: "SELECT id, name FROM users ORDER BY id", Parameters: nil, Columns: engineRecord.ReturnedCols},
+			},
+		}, nil
+	}
+	conf, _ := config.ParseConfig(strings.NewReader(testPluginPipelineConfig))
+	inputs := &sourceFiles{
+		Config: &conf, ConfigPath: "sqlc.yaml", Dir: ".",
+		FileContents: map[string][]byte{"schema.sql": []byte(schemaContent), "query.sql": []byte(queryContent)},
+	}
+	debug := opts.DebugFromString("")
+	debug.ProcessPlugins = true
+	o := &Options{
+		Env: Env{Debug: debug}, Stderr: &bytes.Buffer{}, PluginParseFunc: pluginParse,
+		CodegenHandlerOverride: ext.HandleFunc(func(_ context.Context, req *plugin.GenerateRequest) (*plugin.GenerateResponse, error) {
+			codegenReq = req
+			return &plugin.GenerateResponse{}, nil
+		}),
+	}
+	_, err := generate(context.Background(), inputs, o)
+	if err != nil {
+		t.Fatalf("generate failed: %v", err)
+	}
+	if engineRecord.Calls != 1 {
+		t.Errorf("expected 1 Parse call (full query.sql), got %d", engineRecord.Calls)
+	}
+	if len(engineRecord.CalledWith) != 1 {
+		t.Fatalf("expected 1 CalledWith, got %d", len(engineRecord.CalledWith))
+	}
+	if engineRecord.CalledWith[0].SchemaSQL != schemaContent {
+		t.Errorf("Parse must receive schema; got %q", engineRecord.CalledWith[0].SchemaSQL)
+	}
+	if engineRecord.CalledWith[0].QuerySQL != queryContent {
+		t.Errorf("Parse must receive full query.sql; got %q", engineRecord.CalledWith[0].QuerySQL)
+	}
+	if codegenReq == nil || len(codegenReq.Queries) != 2 {
+		t.Errorf("codegen must receive 2 queries (from 2 statements); got %d", len(codegenReq.Queries))
+	}
+}
+
+const testPluginPipelineConfigDatabaseOnly = `version: "2"
+sql:
+  - engine: "testeng"
+    schema: ["schema.sql"]
+    queries: ["query.sql"]
+    analyzer:
+      database: only
+    database:
+      uri: "postgres://localhost/test"
+    codegen:
+      - plugin: "mock"
+        out: "."
+plugins:
+  - name: "mock"
+    process:
+      cmd: "echo"
+engines:
+  - name: "testeng"
+    process:
+      cmd: "echo"
+`
+
+// TestPluginPipeline_DatabaseOnly_ReceivesNoSchema verifies that in databaseOnly mode (analyzer.database: only +
+// database.uri) the plugin receives empty schema and the core would pass connection_params to the real runner.
+// The mock only sees (schemaSQL, querySQL); in databaseOnly we pass schemaSQL="".
+func TestPluginPipeline_DatabaseOnly_ReceivesNoSchema(t *testing.T) {
+	const queryContent = "-- name: GetOne :one\nSELECT 1"
+	engineRecord := &engineMockRecord{
+		ReturnedSQL: "SELECT 1", ReturnedParams: nil, ReturnedCols: []*pb.Column{{Name: "?column?", DataType: "int", Nullable: true}},
+	}
+	pluginParse := func(schemaSQL, querySQL string) (*pb.ParseResponse, error) {
+		engineRecord.Calls++
+		engineRecord.CalledWith = append(engineRecord.CalledWith, struct{ SchemaSQL, QuerySQL string }{schemaSQL, querySQL})
+		return &pb.ParseResponse{
+			Statements: []*pb.Statement{{Name: "GetOne", Cmd: pb.Cmd_CMD_ONE, Sql: "SELECT 1", Parameters: nil, Columns: engineRecord.ReturnedCols}},
+		}, nil
+	}
+	conf, err := config.ParseConfig(strings.NewReader(testPluginPipelineConfigDatabaseOnly))
+	if err != nil {
+		t.Fatalf("parse config: %v", err)
+	}
+	inputs := &sourceFiles{
+		Config: &conf, ConfigPath: "sqlc.yaml", Dir: ".",
+		FileContents: map[string][]byte{"schema.sql": []byte("CREATE TABLE t (id INT);"), "query.sql": []byte(queryContent)},
+	}
+	debug := opts.DebugFromString("")
+	debug.ProcessPlugins = true
+	o := &Options{
+		Env: Env{Debug: debug}, Stderr: &bytes.Buffer{}, PluginParseFunc: pluginParse,
+		CodegenHandlerOverride: ext.HandleFunc(func(_ context.Context, req *plugin.GenerateRequest) (*plugin.GenerateResponse, error) { return &plugin.GenerateResponse{}, nil }),
+	}
+	_, err = generate(context.Background(), inputs, o)
+	if err != nil {
+		t.Fatalf("generate failed: %v", err)
+	}
+	if len(engineRecord.CalledWith) != 1 {
+		t.Errorf("expected 1 Parse call, got %d", len(engineRecord.CalledWith))
+	}
+	if len(engineRecord.CalledWith) > 0 && engineRecord.CalledWith[0].SchemaSQL != "" {
+		t.Errorf("databaseOnly mode: each Parse call must receive empty schema (connection_params are used by real runner); got %q", engineRecord.CalledWith[0].SchemaSQL)
+	}
+	if len(engineRecord.CalledWith) > 0 && engineRecord.CalledWith[0].QuerySQL != queryContent {
+		t.Errorf("query SQL must still be passed; got %q", engineRecord.CalledWith[0].QuerySQL)
+	}
+}
+
+// TestPluginPipeline_OptionsOverrideNil ensures default Options do not inject mocks.
+func TestPluginPipeline_OptionsOverrideNil(t *testing.T) {
+	o := &Options{}
+	if o.CodegenHandlerOverride != nil || o.PluginParseFunc != nil {
+		t.Error("default Options should have nil overrides")
+	}
+}
diff --git a/internal/cmd/process.go b/internal/cmd/process.go
index 5003d113b8..03c6028504 100644
--- a/internal/cmd/process.go
+++ b/internal/cmd/process.go
@@ -8,28 +8,30 @@ import (
 	"path/filepath"
 	"runtime"
 	"runtime/trace"
+	"strings"
 
 	"golang.org/x/sync/errgroup"
 
 	"github.com/sqlc-dev/sqlc/internal/compiler"
 	"github.com/sqlc-dev/sqlc/internal/config"
 	"github.com/sqlc-dev/sqlc/internal/debug"
+	"github.com/sqlc-dev/sqlc/internal/multierr"
 	"github.com/sqlc-dev/sqlc/internal/opts"
 )
 
-type OutputPair struct {
+type outputPair struct {
 	Gen    config.SQLGen
 	Plugin *config.Codegen
 
 	config.SQL
 }
 
-type ResultProcessor interface {
-	Pairs(context.Context, *config.Config) []OutputPair
-	ProcessResult(context.Context, config.CombinedSettings, OutputPair, *compiler.Result) error
+type resultProcessor interface {
+	Pairs(context.Context, *config.Config) []outputPair
+	ProcessResult(context.Context, config.CombinedSettings, outputPair, *compiler.Result) error
 }
 
-func Process(ctx context.Context, rp ResultProcessor, dir, filename string, o *Options) error {
+func Process(ctx context.Context, rp resultProcessor, dir, filename string, o *Options) error {
 	e := o.Env
 	stderr := o.Stderr
 
@@ -49,11 +51,14 @@ func Process(ctx context.Context, rp ResultProcessor, dir, filename string, o *O
 		return err
 	}
 
-	return processQuerySets(ctx, rp, conf, dir, o)
+	inputs := &sourceFiles{Config: conf, ConfigPath: configPath, Dir: dir}
+	return processQuerySets(ctx, rp, inputs, o)
 }
 
-func processQuerySets(ctx context.Context, rp ResultProcessor, conf *config.Config, dir string, o *Options) error {
+func processQuerySets(ctx context.Context, rp resultProcessor, inputs *sourceFiles, o *Options) error {
 	stderr := o.Stderr
+	conf := inputs.Config
+	dir := inputs.Dir
 
 	errored := false
 
@@ -69,6 +74,9 @@ func processQuerySets(ctx context.Context, rp ResultProcessor, conf *config.Conf
 
 		grp.Go(func() error {
 			combo := config.Combine(*conf, sql.SQL)
+			if dir != "" {
+				combo.Dir = dir
+			}
 			if sql.Plugin != nil {
 				combo.Codegen = *sql.Plugin
 			}
@@ -104,16 +112,36 @@ func processQuerySets(ctx context.Context, rp ResultProcessor, conf *config.Conf
 			packageRegion := trace.StartRegion(gctx, "package")
 			trace.Logf(gctx, "", "name=%s dir=%s plugin=%s", name, dir, lang)
 
-			result, failed := parse(gctx, name, dir, sql.SQL, combo, parseOpts, errout)
-			if failed {
-				packageRegion.End()
-				errored = true
-				return nil
-			}
-			if err := rp.ProcessResult(gctx, combo, sql, result); err != nil {
-				fmt.Fprintf(errout, "# package %s\n", name)
-				fmt.Fprintf(errout, "error generating code: %s\n", err)
-				errored = true
+			if !config.IsBuiltinEngine(combo.Package.Engine) {
+				// Plugin engine: skip compiler and AST; call engine plugin with schema + query, then codegen.
+				if err := runPluginQuerySet(gctx, rp, name, dir, sql, combo, inputs, o); err != nil {
+					// Use the same format as the compiler path for unknown-engine errors (for backward compatibility).
+					if strings.HasPrefix(err.Error(), "unknown engine:") {
+						fmt.Fprintf(errout, "error creating compiler: %v\n", err)
+					} else {
+						fmt.Fprintf(errout, "# package %s\n", name)
+						if multi, ok := err.(*multierr.Error); ok {
+							for _, e := range multi.Errs() {
+								fmt.Fprintf(errout, "%s:%d:%d: %v\n", e.Filename, e.Line, e.Column, e.Err)
+							}
+						} else {
+							fmt.Fprintf(errout, "%v\n", err)
+						}
+					}
+					errored = true
+				}
+			} else {
+				result, failed := parse(gctx, name, dir, sql.SQL, combo, parseOpts, errout)
+				if failed {
+					packageRegion.End()
+					errored = true
+					return nil
+				}
+				if err := rp.ProcessResult(gctx, combo, sql, result); err != nil {
+					fmt.Fprintf(errout, "# package %s\n", name)
+					fmt.Fprintf(errout, "error generating code: %s\n", err)
+					errored = true
+				}
 			}
 			packageRegion.End()
 			return nil
diff --git a/internal/cmd/push.go b/internal/cmd/push.go
index 19aca6739b..5461c47da2 100644
--- a/internal/cmd/push.go
+++ b/internal/cmd/push.go
@@ -46,17 +46,17 @@ type pusher struct {
 	results []*bundler.QuerySetArchive
 }
 
-func (g *pusher) Pairs(ctx context.Context, conf *config.Config) []OutputPair {
-	var pairs []OutputPair
+func (g *pusher) Pairs(ctx context.Context, conf *config.Config) []outputPair {
+	var pairs []outputPair
 	for _, sql := range conf.SQL {
-		pairs = append(pairs, OutputPair{
+		pairs = append(pairs, outputPair{
 			SQL: sql,
 		})
 	}
 	return pairs
 }
 
-func (g *pusher) ProcessResult(ctx context.Context, combo config.CombinedSettings, sql OutputPair, result *compiler.Result) error {
+func (g *pusher) ProcessResult(ctx context.Context, combo config.CombinedSettings, sql outputPair, result *compiler.Result) error {
 	req := codeGenRequest(result, combo)
 	g.m.Lock()
 	g.results = append(g.results, &bundler.QuerySetArchive{
diff --git a/internal/cmd/shim.go b/internal/cmd/shim.go
index 654500429a..648ad7a2ab 100644
--- a/internal/cmd/shim.go
+++ b/internal/cmd/shim.go
@@ -224,9 +224,13 @@ func pluginQueryParam(p compiler.Parameter) *plugin.Parameter {
 }
 
 func codeGenRequest(r *compiler.Result, settings config.CombinedSettings) *plugin.GenerateRequest {
+	cat := pluginCatalog(r.Catalog)
+	if r.PluginCatalog != nil {
+		cat = r.PluginCatalog
+	}
 	return &plugin.GenerateRequest{
 		Settings:    pluginSettings(r, settings),
-		Catalog:     pluginCatalog(r.Catalog),
+		Catalog:     cat,
 		Queries:     pluginQueries(r),
 		SqlcVersion: info.Version,
 	}
diff --git a/internal/compiler/result.go b/internal/compiler/result.go
index 3647da630f..7c1ec39d8b 100644
--- a/internal/compiler/result.go
+++ b/internal/compiler/result.go
@@ -1,10 +1,14 @@
 package compiler
 
 import (
+	"github.com/sqlc-dev/sqlc/internal/plugin"
 	"github.com/sqlc-dev/sqlc/internal/sql/catalog"
 )
 
 type Result struct {
-	Catalog *catalog.Catalog
-	Queries []*Query
+	Catalog      *catalog.Catalog
+	Queries      []*Query
+	// PluginCatalog is set by plugin engines (e.g. sqlc-engine-ydb) from ParseResponse.Catalog.
+	// When non-nil, it is passed to codegen instead of converting Catalog.
+	PluginCatalog *plugin.Catalog
 }
diff --git a/internal/config/config.go b/internal/config/config.go
index d3e610ef05..7d6153f26b 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -63,10 +63,43 @@ type Config struct {
 	SQL       []SQL                `json:"sql" yaml:"sql"`
 	Overrides Overrides            `json:"overrides,omitempty" yaml:"overrides"`
 	Plugins   []Plugin             `json:"plugins" yaml:"plugins"`
+	Engines   []EnginePlugin       `json:"engines" yaml:"engines"`
 	Rules     []Rule               `json:"rules" yaml:"rules"`
 	Options   map[string]yaml.Node `json:"options" yaml:"options"`
 }
 
+// EnginePlugin defines a custom database engine plugin.
+// Engine plugins allow external SQL parsers and database backends to be used with sqlc.
+type EnginePlugin struct {
+	// Name is the unique name for this engine (used in sql[].engine field)
+	Name string `json:"name" yaml:"name"`
+
+	// Env is a list of environment variable names to pass to the plugin
+	Env []string `json:"env" yaml:"env"`
+
+	// Process defines an engine plugin that runs as an external process
+	Process *EnginePluginProcess `json:"process" yaml:"process"`
+
+	// WASM defines an engine plugin that runs as a WASM module
+	WASM *EnginePluginWASM `json:"wasm" yaml:"wasm"`
+}
+
+// EnginePluginProcess defines a process-based engine plugin.
+type EnginePluginProcess struct {
+	// Cmd is the command to run (must be in PATH or an absolute path)
+	Cmd string `json:"cmd" yaml:"cmd"`
+}
+
+// EnginePluginWASM defines a WASM-based engine plugin.
+type EnginePluginWASM struct {
+	// URL is the URL to download the WASM module from
+	// Supports file:// and https:// schemes
+	URL string `json:"url" yaml:"url"`
+
+	// SHA256 is the expected SHA256 checksum of the WASM module
+	SHA256 string `json:"sha256" yaml:"sha256"`
+}
+
 type Server struct {
 	Name   string `json:"name,omitempty" yaml:"name"`
 	Engine Engine `json:"engine,omitempty" yaml:"engine"`
@@ -125,8 +158,8 @@ type SQL struct {
 // AnalyzerDatabase represents the database analyzer setting.
 // It can be a boolean (true/false) or the string "only" for database-only mode.
 type AnalyzerDatabase struct {
-	value   *bool  // nil means not set, true/false for boolean values
-	isOnly  bool   // true when set to "only"
+	value  *bool // nil means not set, true/false for boolean values
+	isOnly bool  // true when set to "only"
 }
 
 // IsEnabled returns true if the database analyzer should be used.
@@ -228,6 +261,14 @@ var ErrPluginNoType = errors.New("plugin: field `process` or `wasm` required")
 var ErrPluginBothTypes = errors.New("plugin: `process` and `wasm` cannot both be defined")
 var ErrPluginProcessNoCmd = errors.New("plugin: missing process command")
 
+var ErrEnginePluginNoName = errors.New("engine plugin: missing name")
+var ErrEnginePluginBuiltin = errors.New("engine plugin: cannot override built-in engine")
+var ErrEnginePluginExists = errors.New("engine plugin: a plugin with that name already exists")
+var ErrEnginePluginNoType = errors.New("engine plugin: field `process` or `wasm` required")
+var ErrEnginePluginBothTypes = errors.New("engine plugin: `process` and `wasm` cannot both be defined")
+var ErrEnginePluginProcessNoCmd = errors.New("engine plugin: missing process command")
+var ErrEnginePluginWASMNoURL = errors.New("engine plugin: missing wasm url")
+
 var ErrInvalidDatabase = errors.New("database must be managed or have a non-empty URI")
 var ErrManagedDatabaseNoProject = errors.New(`managed databases require a cloud project
 
@@ -285,6 +326,9 @@ type CombinedSettings struct {
 
 	// TODO: Combine these into a more usable type
 	Codegen Codegen
+
+	// Dir is the directory containing the config file (for resolving relative paths)
+	Dir string
 }
 
 func Combine(conf Config, pkg SQL) CombinedSettings {
diff --git a/internal/config/validate.go b/internal/config/validate.go
index fadef4fb3b..6587283ea3 100644
--- a/internal/config/validate.go
+++ b/internal/config/validate.go
@@ -1,6 +1,46 @@
 package config
 
+// builtinEngines contains the names of built-in database engines.
+var builtinEngines = map[Engine]bool{
+	EngineMySQL:      true,
+	EnginePostgreSQL: true,
+	EngineSQLite:     true,
+}
+
+// IsBuiltinEngine returns true if the engine name is a built-in engine.
+func IsBuiltinEngine(name Engine) bool {
+	return builtinEngines[name]
+}
+
 func Validate(c *Config) error {
+	// Validate engine plugins
+	engineNames := make(map[string]bool)
+	for _, ep := range c.Engines {
+		if ep.Name == "" {
+			return ErrEnginePluginNoName
+		}
+		if IsBuiltinEngine(Engine(ep.Name)) {
+			return ErrEnginePluginBuiltin
+		}
+		if engineNames[ep.Name] {
+			return ErrEnginePluginExists
+		}
+		engineNames[ep.Name] = true
+
+		if ep.Process == nil && ep.WASM == nil {
+			return ErrEnginePluginNoType
+		}
+		if ep.Process != nil && ep.WASM != nil {
+			return ErrEnginePluginBothTypes
+		}
+		if ep.Process != nil && ep.Process.Cmd == "" {
+			return ErrEnginePluginProcessNoCmd
+		}
+		if ep.WASM != nil && ep.WASM.URL == "" {
+			return ErrEnginePluginWASMNoURL
+		}
+	}
+
 	for _, sql := range c.SQL {
 		if sql.Database != nil {
 			if sql.Database.URI == "" && !sql.Database.Managed {
@@ -10,3 +50,13 @@ func Validate(c *Config) error {
 	}
 	return nil
 }
+
+// FindEnginePlugin finds an engine plugin by name.
+func FindEnginePlugin(c *Config, name string) (*EnginePlugin, bool) {
+	for i := range c.Engines {
+		if c.Engines[i].Name == name {
+			return &c.Engines[i], true
+		}
+	}
+	return nil, false
+}
diff --git a/internal/endtoend/testdata/bad_config/engine/stderr.txt b/internal/endtoend/testdata/bad_config/engine/stderr.txt
index 9797244924..f5f83ed3c2 100644
--- a/internal/endtoend/testdata/bad_config/engine/stderr.txt
+++ b/internal/endtoend/testdata/bad_config/engine/stderr.txt
@@ -1 +1,3 @@
-error creating compiler: unknown engine: bad_engine
\ No newline at end of file
+error creating compiler: unknown engine: bad_engine
+
+Add the engine to the 'engines' section of sqlc.yaml. See the engine plugins documentation: https://docs.sqlc.dev/en/latest/guides/engine-plugins.html
diff --git a/internal/metadata/meta.go b/internal/metadata/meta.go
index 8f63624d2c..d0bcf76fa6 100644
--- a/internal/metadata/meta.go
+++ b/internal/metadata/meta.go
@@ -118,6 +118,86 @@ func ParseQueryNameAndType(t string, commentStyle CommentSyntax) (string, string
 	return "", "", nil
 }
 
+// QueryBlock is one named query block (from " name: X :cmd" to the next such line or EOF).
+type QueryBlock struct {
+	SQL  string
+	Name string
+	Cmd  string
+}
+
+// isBlockStartLine reports whether the line starts a named query block (e.g. "-- name: GetUser :one").
+func isBlockStartLine(line string, commentStyle CommentSyntax) bool {
+	line = strings.TrimSpace(line)
+	var rest string
+	switch {
+	case strings.HasPrefix(line, "--"):
+		if !commentStyle.Dash {
+			return false
+		}
+		rest = line[2:] // keep " name: X :cmd" with leading space to match ParseQueryNameAndType
+	case strings.HasPrefix(line, "#"):
+		if !commentStyle.Hash {
+			return false
+		}
+		rest = line[1:]
+	case strings.HasPrefix(line, "/*"):
+		if !commentStyle.SlashStar {
+			return false
+		}
+		rest = line[2:]
+		if strings.HasSuffix(rest, "*/") {
+			rest = rest[:len(rest)-2]
+		}
+		rest = strings.TrimSpace(rest)
+	default:
+		return false
+	}
+	if !strings.HasPrefix(rest, " name: ") {
+		return false
+	}
+	part := strings.Split(rest, " ")
+	if len(part) < 4 {
+		return false
+	}
+	queryType := strings.TrimSpace(part[3])
+	switch queryType {
+	case CmdOne, CmdMany, CmdExec, CmdExecResult, CmdExecRows, CmdExecLastId, CmdCopyFrom, CmdBatchExec, CmdBatchMany, CmdBatchOne:
+		return true
+	}
+	return false
+}
+
+// QueryBlocks splits content into named query blocks. Each block runs from a " name: X :cmd" line
+// to the next such line (or EOF). Returns one entry per block with non-empty name.
+func QueryBlocks(content string, commentStyle CommentSyntax) ([]QueryBlock, error) {
+	lines := strings.Split(content, "\n")
+	var starts []int
+	pos := 0
+	for _, line := range lines {
+		if isBlockStartLine(line, commentStyle) {
+			starts = append(starts, pos)
+		}
+		pos += len(line) + 1
+	}
+	var out []QueryBlock
+	for i := 0; i < len(starts); i++ {
+		end := len(content)
+		if i+1 < len(starts) {
+			end = starts[i+1]
+		}
+		blockSQL := content[starts[i]:end]
+		name, cmd, err := ParseQueryNameAndType(blockSQL, commentStyle)
+		if err != nil {
+			return nil, err
+		}
+		if name == "" {
+			continue
+		}
+		out = append(out, QueryBlock{SQL: blockSQL, Name: name, Cmd: cmd})
+	}
+	return out, nil
+}
+
 // ParseCommentFlags processes the comments provided with queries to determine the metadata params, flags and rules to skip.
 // All flags in query comments are prefixed with `@`, e.g. @param, @@sqlc-vet-disable.
 func ParseCommentFlags(comments []string) (map[string]string, map[string]bool, map[string]struct{}, error) {
diff --git a/pkg/engine/engine.pb.go b/pkg/engine/engine.pb.go
new file mode 100644
index 0000000000..a9f4982995
--- /dev/null
+++ b/pkg/engine/engine.pb.go
@@ -0,0 +1,1101 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.36.11
+// 	protoc        v6.32.1
+// source: protos/engine/engine.proto
+
+package engine
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+	unsafe "unsafe"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// Cmd is the query command/type. Matches sqlc's ":one", ":many", ":exec", etc.
+type Cmd int32
+
+const (
+	Cmd_CMD_UNSPECIFIED  Cmd = 0
+	Cmd_CMD_ONE          Cmd = 1
+	Cmd_CMD_MANY         Cmd = 2
+	Cmd_CMD_EXEC         Cmd = 3
+	Cmd_CMD_EXEC_RESULT  Cmd = 4
+	Cmd_CMD_EXEC_ROWS    Cmd = 5
+	Cmd_CMD_EXEC_LAST_ID Cmd = 6
+	Cmd_CMD_COPY_FROM    Cmd = 7
+	Cmd_CMD_BATCH_EXEC   Cmd = 8
+	Cmd_CMD_BATCH_MANY   Cmd = 9
+	Cmd_CMD_BATCH_ONE    Cmd = 10
+)
+
+// Enum value maps for Cmd.
+var (
+	Cmd_name = map[int32]string{
+		0:  "CMD_UNSPECIFIED",
+		1:  "CMD_ONE",
+		2:  "CMD_MANY",
+		3:  "CMD_EXEC",
+		4:  "CMD_EXEC_RESULT",
+		5:  "CMD_EXEC_ROWS",
+		6:  "CMD_EXEC_LAST_ID",
+		7:  "CMD_COPY_FROM",
+		8:  "CMD_BATCH_EXEC",
+		9:  "CMD_BATCH_MANY",
+		10: "CMD_BATCH_ONE",
+	}
+	Cmd_value = map[string]int32{
+		"CMD_UNSPECIFIED":  0,
+		"CMD_ONE":          1,
+		"CMD_MANY":         2,
+		"CMD_EXEC":         3,
+		"CMD_EXEC_RESULT":  4,
+		"CMD_EXEC_ROWS":    5,
+		"CMD_EXEC_LAST_ID": 6,
+		"CMD_COPY_FROM":    7,
+		"CMD_BATCH_EXEC":   8,
+		"CMD_BATCH_MANY":   9,
+		"CMD_BATCH_ONE":    10,
+	}
+)
+
+func (x Cmd) Enum() *Cmd {
+	p := new(Cmd)
+	*p = x
+	return p
+}
+
+func (x Cmd) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (Cmd) Descriptor() protoreflect.EnumDescriptor {
+	return file_protos_engine_engine_proto_enumTypes[0].Descriptor()
+}
+
+func (Cmd) Type() protoreflect.EnumType {
+	return &file_protos_engine_engine_proto_enumTypes[0]
+}
+
+func (x Cmd) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use Cmd.Descriptor instead.
+func (Cmd) EnumDescriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{0}
+}
+
+// ParseRequest contains the SQL to parse and either schema or connection parameters.
+type ParseRequest struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// The SQL query text to parse.
+	Sql string `protobuf:"bytes,1,opt,name=sql,proto3" json:"sql,omitempty"`
+	// Either schema SQL or connection parameters for database-only mode.
+	//
+	// Types that are valid to be assigned to SchemaSource:
+	//
+	//	*ParseRequest_SchemaSql
+	//	*ParseRequest_ConnectionParams
+	SchemaSource  isParseRequest_SchemaSource `protobuf_oneof:"schema_source"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *ParseRequest) Reset() {
+	*x = ParseRequest{}
+	mi := &file_protos_engine_engine_proto_msgTypes[0]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *ParseRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ParseRequest) ProtoMessage() {}
+
+func (x *ParseRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[0]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ParseRequest.ProtoReflect.Descriptor instead.
+func (*ParseRequest) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *ParseRequest) GetSql() string {
+	if x != nil {
+		return x.Sql
+	}
+	return ""
+}
+
+func (x *ParseRequest) GetSchemaSource() isParseRequest_SchemaSource {
+	if x != nil {
+		return x.SchemaSource
+	}
+	return nil
+}
+
+func (x *ParseRequest) GetSchemaSql() string {
+	if x != nil {
+		if x, ok := x.SchemaSource.(*ParseRequest_SchemaSql); ok {
+			return x.SchemaSql
+		}
+	}
+	return ""
+}
+
+func (x *ParseRequest) GetConnectionParams() *ConnectionParams {
+	if x != nil {
+		if x, ok := x.SchemaSource.(*ParseRequest_ConnectionParams); ok {
+			return x.ConnectionParams
+		}
+	}
+	return nil
+}
+
+type isParseRequest_SchemaSource interface {
+	isParseRequest_SchemaSource()
+}
+
+type ParseRequest_SchemaSql struct {
+	// Schema SQL text (schema.sql) for schema-based parsing.
+	SchemaSql string `protobuf:"bytes,2,opt,name=schema_sql,json=schemaSql,proto3,oneof"`
+}
+
+type ParseRequest_ConnectionParams struct {
+	// Connection parameters for database-only mode.
+	ConnectionParams *ConnectionParams `protobuf:"bytes,3,opt,name=connection_params,json=connectionParams,proto3,oneof"`
+}
+
+func (*ParseRequest_SchemaSql) isParseRequest_SchemaSource() {}
+
+func (*ParseRequest_ConnectionParams) isParseRequest_SchemaSource() {}
+
+// ConnectionParams contains database connection parameters for database-only mode.
+type ConnectionParams struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// Database connection string or DSN.
+	Dsn string `protobuf:"bytes,1,opt,name=dsn,proto3" json:"dsn,omitempty"`
+	// Additional connection parameters as key-value pairs.
+	Params        map[string]string `protobuf:"bytes,2,rep,name=params,proto3" json:"params,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *ConnectionParams) Reset() {
+	*x = ConnectionParams{}
+	mi := &file_protos_engine_engine_proto_msgTypes[1]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *ConnectionParams) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ConnectionParams) ProtoMessage() {}
+
+func (x *ConnectionParams) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[1]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ConnectionParams.ProtoReflect.Descriptor instead.
+func (*ConnectionParams) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *ConnectionParams) GetDsn() string {
+	if x != nil {
+		return x.Dsn
+	}
+	return ""
+}
+
+func (x *ConnectionParams) GetParams() map[string]string {
+	if x != nil {
+		return x.Params
+	}
+	return nil
+}
+
+// Statement is one parsed query block: name/cmd from " name: X :cmd", plus sql and type info.
+type Statement struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// Query name (from "-- name: GetUser" etc.).
+	Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	// Command/type for this statement.
+	Cmd Cmd `protobuf:"varint,2,opt,name=cmd,proto3,enum=engine.Cmd" json:"cmd,omitempty"`
+	// Processed SQL for this statement (wildcards expanded if applicable).
+	Sql string `protobuf:"bytes,3,opt,name=sql,proto3" json:"sql,omitempty"`
+	// Parameters for this statement.
+	Parameters []*Parameter `protobuf:"bytes,4,rep,name=parameters,proto3" json:"parameters,omitempty"`
+	// Result columns for this statement.
+	Columns       []*Column `protobuf:"bytes,5,rep,name=columns,proto3" json:"columns,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *Statement) Reset() {
+	*x = Statement{}
+	mi := &file_protos_engine_engine_proto_msgTypes[2]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Statement) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Statement) ProtoMessage() {}
+
+func (x *Statement) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[2]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Statement.ProtoReflect.Descriptor instead.
+func (*Statement) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *Statement) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *Statement) GetCmd() Cmd {
+	if x != nil {
+		return x.Cmd
+	}
+	return Cmd_CMD_UNSPECIFIED
+}
+
+func (x *Statement) GetSql() string {
+	if x != nil {
+		return x.Sql
+	}
+	return ""
+}
+
+func (x *Statement) GetParameters() []*Parameter {
+	if x != nil {
+		return x.Parameters
+	}
+	return nil
+}
+
+func (x *Statement) GetColumns() []*Column {
+	if x != nil {
+		return x.Columns
+	}
+	return nil
+}
+
+// ParseResponse contains the processed statements and optional catalog. The plugin receives
+// the full query file and schema (or connection params); it returns one Statement per query
+// block. Catalog is derived from schema_sql or connection_params and is passed to codegen.
+type ParseResponse struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Statements    []*Statement           `protobuf:"bytes,1,rep,name=statements,proto3" json:"statements,omitempty"`
+	Catalog       *Catalog               `protobuf:"bytes,2,opt,name=catalog,proto3" json:"catalog,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *ParseResponse) Reset() {
+	*x = ParseResponse{}
+	mi := &file_protos_engine_engine_proto_msgTypes[3]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *ParseResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ParseResponse) ProtoMessage() {}
+
+func (x *ParseResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[3]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ParseResponse.ProtoReflect.Descriptor instead.
+func (*ParseResponse) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *ParseResponse) GetStatements() []*Statement {
+	if x != nil {
+		return x.Statements
+	}
+	return nil
+}
+
+func (x *ParseResponse) GetCatalog() *Catalog {
+	if x != nil {
+		return x.Catalog
+	}
+	return nil
+}
+
+// Identifier identifies a catalog/schema/name (used in Catalog types).
+type Identifier struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Catalog       string                 `protobuf:"bytes,1,opt,name=catalog,proto3" json:"catalog,omitempty"`
+	Schema        string                 `protobuf:"bytes,2,opt,name=schema,proto3" json:"schema,omitempty"`
+	Name          string                 `protobuf:"bytes,3,opt,name=name,proto3" json:"name,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *Identifier) Reset() {
+	*x = Identifier{}
+	mi := &file_protos_engine_engine_proto_msgTypes[4]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Identifier) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Identifier) ProtoMessage() {}
+
+func (x *Identifier) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[4]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Identifier.ProtoReflect.Descriptor instead.
+func (*Identifier) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{4}
+}
+
+func (x *Identifier) GetCatalog() string {
+	if x != nil {
+		return x.Catalog
+	}
+	return ""
+}
+
+func (x *Identifier) GetSchema() string {
+	if x != nil {
+		return x.Schema
+	}
+	return ""
+}
+
+func (x *Identifier) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+// Catalog mirrors plugin.Catalog so engine plugins can return schema without importing plugin.
+// The sqlc app converts it to plugin.Catalog when calling codegen.
+type Catalog struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Comment       string                 `protobuf:"bytes,1,opt,name=comment,proto3" json:"comment,omitempty"`
+	DefaultSchema string                 `protobuf:"bytes,2,opt,name=default_schema,json=defaultSchema,proto3" json:"default_schema,omitempty"`
+	Name          string                 `protobuf:"bytes,3,opt,name=name,proto3" json:"name,omitempty"`
+	Schemas       []*CatalogSchema       `protobuf:"bytes,4,rep,name=schemas,proto3" json:"schemas,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *Catalog) Reset() {
+	*x = Catalog{}
+	mi := &file_protos_engine_engine_proto_msgTypes[5]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Catalog) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Catalog) ProtoMessage() {}
+
+func (x *Catalog) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[5]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Catalog.ProtoReflect.Descriptor instead.
+func (*Catalog) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{5}
+}
+
+func (x *Catalog) GetComment() string {
+	if x != nil {
+		return x.Comment
+	}
+	return ""
+}
+
+func (x *Catalog) GetDefaultSchema() string {
+	if x != nil {
+		return x.DefaultSchema
+	}
+	return ""
+}
+
+func (x *Catalog) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *Catalog) GetSchemas() []*CatalogSchema {
+	if x != nil {
+		return x.Schemas
+	}
+	return nil
+}
+
+type CatalogSchema struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Comment       string                 `protobuf:"bytes,1,opt,name=comment,proto3" json:"comment,omitempty"`
+	Name          string                 `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"`
+	Tables        []*CatalogTable        `protobuf:"bytes,3,rep,name=tables,proto3" json:"tables,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *CatalogSchema) Reset() {
+	*x = CatalogSchema{}
+	mi := &file_protos_engine_engine_proto_msgTypes[6]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *CatalogSchema) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CatalogSchema) ProtoMessage() {}
+
+func (x *CatalogSchema) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[6]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CatalogSchema.ProtoReflect.Descriptor instead.
+func (*CatalogSchema) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{6}
+}
+
+func (x *CatalogSchema) GetComment() string {
+	if x != nil {
+		return x.Comment
+	}
+	return ""
+}
+
+func (x *CatalogSchema) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *CatalogSchema) GetTables() []*CatalogTable {
+	if x != nil {
+		return x.Tables
+	}
+	return nil
+}
+
+type CatalogTable struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Rel           *Identifier            `protobuf:"bytes,1,opt,name=rel,proto3" json:"rel,omitempty"`
+	Columns       []*CatalogColumn       `protobuf:"bytes,2,rep,name=columns,proto3" json:"columns,omitempty"`
+	Comment       string                 `protobuf:"bytes,3,opt,name=comment,proto3" json:"comment,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *CatalogTable) Reset() {
+	*x = CatalogTable{}
+	mi := &file_protos_engine_engine_proto_msgTypes[7]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *CatalogTable) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CatalogTable) ProtoMessage() {}
+
+func (x *CatalogTable) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[7]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CatalogTable.ProtoReflect.Descriptor instead.
+func (*CatalogTable) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{7}
+}
+
+func (x *CatalogTable) GetRel() *Identifier {
+	if x != nil {
+		return x.Rel
+	}
+	return nil
+}
+
+func (x *CatalogTable) GetColumns() []*CatalogColumn {
+	if x != nil {
+		return x.Columns
+	}
+	return nil
+}
+
+func (x *CatalogTable) GetComment() string {
+	if x != nil {
+		return x.Comment
+	}
+	return ""
+}
+
+type CatalogColumn struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Name          string                 `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	NotNull       bool                   `protobuf:"varint,3,opt,name=not_null,json=notNull,proto3" json:"not_null,omitempty"`
+	IsArray       bool                   `protobuf:"varint,4,opt,name=is_array,json=isArray,proto3" json:"is_array,omitempty"`
+	Comment       string                 `protobuf:"bytes,5,opt,name=comment,proto3" json:"comment,omitempty"`
+	Length        int32                  `protobuf:"varint,6,opt,name=length,proto3" json:"length,omitempty"`
+	ArrayDims     int32                  `protobuf:"varint,17,opt,name=array_dims,json=arrayDims,proto3" json:"array_dims,omitempty"`
+	Type          *Identifier            `protobuf:"bytes,12,opt,name=type,proto3" json:"type,omitempty"`
+	Table         *Identifier            `protobuf:"bytes,10,opt,name=table,proto3" json:"table,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *CatalogColumn) Reset() {
+	*x = CatalogColumn{}
+	mi := &file_protos_engine_engine_proto_msgTypes[8]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *CatalogColumn) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CatalogColumn) ProtoMessage() {}
+
+func (x *CatalogColumn) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[8]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CatalogColumn.ProtoReflect.Descriptor instead.
+func (*CatalogColumn) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{8}
+}
+
+func (x *CatalogColumn) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *CatalogColumn) GetNotNull() bool {
+	if x != nil {
+		return x.NotNull
+	}
+	return false
+}
+
+func (x *CatalogColumn) GetIsArray() bool {
+	if x != nil {
+		return x.IsArray
+	}
+	return false
+}
+
+func (x *CatalogColumn) GetComment() string {
+	if x != nil {
+		return x.Comment
+	}
+	return ""
+}
+
+func (x *CatalogColumn) GetLength() int32 {
+	if x != nil {
+		return x.Length
+	}
+	return 0
+}
+
+func (x *CatalogColumn) GetArrayDims() int32 {
+	if x != nil {
+		return x.ArrayDims
+	}
+	return 0
+}
+
+func (x *CatalogColumn) GetType() *Identifier {
+	if x != nil {
+		return x.Type
+	}
+	return nil
+}
+
+func (x *CatalogColumn) GetTable() *Identifier {
+	if x != nil {
+		return x.Table
+	}
+	return nil
+}
+
+// Parameter represents a query parameter.
+type Parameter struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// Parameter name (if named) or empty for positional parameters.
+	Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	// Parameter position (1-based) for positional parameters.
+	Position int32 `protobuf:"varint,2,opt,name=position,proto3" json:"position,omitempty"`
+	// SQL data type of the parameter.
+	DataType string `protobuf:"bytes,3,opt,name=data_type,json=dataType,proto3" json:"data_type,omitempty"`
+	// Whether the parameter is nullable.
+	Nullable bool `protobuf:"varint,4,opt,name=nullable,proto3" json:"nullable,omitempty"`
+	// Whether the parameter is an array type.
+	IsArray bool `protobuf:"varint,5,opt,name=is_array,json=isArray,proto3" json:"is_array,omitempty"`
+	// Array dimensions if is_array is true.
+	ArrayDims     int32 `protobuf:"varint,6,opt,name=array_dims,json=arrayDims,proto3" json:"array_dims,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *Parameter) Reset() {
+	*x = Parameter{}
+	mi := &file_protos_engine_engine_proto_msgTypes[9]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Parameter) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Parameter) ProtoMessage() {}
+
+func (x *Parameter) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[9]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Parameter.ProtoReflect.Descriptor instead.
+func (*Parameter) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{9}
+}
+
+func (x *Parameter) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *Parameter) GetPosition() int32 {
+	if x != nil {
+		return x.Position
+	}
+	return 0
+}
+
+func (x *Parameter) GetDataType() string {
+	if x != nil {
+		return x.DataType
+	}
+	return ""
+}
+
+func (x *Parameter) GetNullable() bool {
+	if x != nil {
+		return x.Nullable
+	}
+	return false
+}
+
+func (x *Parameter) GetIsArray() bool {
+	if x != nil {
+		return x.IsArray
+	}
+	return false
+}
+
+func (x *Parameter) GetArrayDims() int32 {
+	if x != nil {
+		return x.ArrayDims
+	}
+	return 0
+}
+
+// Column represents a result column.
+type Column struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// Column name.
+	Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	// SQL data type of the column.
+	DataType string `protobuf:"bytes,2,opt,name=data_type,json=dataType,proto3" json:"data_type,omitempty"`
+	// Whether the column is nullable.
+	Nullable bool `protobuf:"varint,3,opt,name=nullable,proto3" json:"nullable,omitempty"`
+	// Whether the column is an array type.
+	IsArray bool `protobuf:"varint,4,opt,name=is_array,json=isArray,proto3" json:"is_array,omitempty"`
+	// Array dimensions if is_array is true.
+	ArrayDims int32 `protobuf:"varint,5,opt,name=array_dims,json=arrayDims,proto3" json:"array_dims,omitempty"`
+	// Table name this column belongs to (if known).
+	TableName string `protobuf:"bytes,6,opt,name=table_name,json=tableName,proto3" json:"table_name,omitempty"`
+	// Schema name this column belongs to (if known).
+	SchemaName    string `protobuf:"bytes,7,opt,name=schema_name,json=schemaName,proto3" json:"schema_name,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *Column) Reset() {
+	*x = Column{}
+	mi := &file_protos_engine_engine_proto_msgTypes[10]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Column) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Column) ProtoMessage() {}
+
+func (x *Column) ProtoReflect() protoreflect.Message {
+	mi := &file_protos_engine_engine_proto_msgTypes[10]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Column.ProtoReflect.Descriptor instead.
+func (*Column) Descriptor() ([]byte, []int) {
+	return file_protos_engine_engine_proto_rawDescGZIP(), []int{10}
+}
+
+func (x *Column) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *Column) GetDataType() string {
+	if x != nil {
+		return x.DataType
+	}
+	return ""
+}
+
+func (x *Column) GetNullable() bool {
+	if x != nil {
+		return x.Nullable
+	}
+	return false
+}
+
+func (x *Column) GetIsArray() bool {
+	if x != nil {
+		return x.IsArray
+	}
+	return false
+}
+
+func (x *Column) GetArrayDims() int32 {
+	if x != nil {
+		return x.ArrayDims
+	}
+	return 0
+}
+
+func (x *Column) GetTableName() string {
+	if x != nil {
+		return x.TableName
+	}
+	return ""
+}
+
+func (x *Column) GetSchemaName() string {
+	if x != nil {
+		return x.SchemaName
+	}
+	return ""
+}
+
+var File_protos_engine_engine_proto protoreflect.FileDescriptor
+
+const file_protos_engine_engine_proto_rawDesc = "" +
+	"\n" +
+	"\x1aprotos/engine/engine.proto\x12\x06engine\"\x9b\x01\n" +
+	"\fParseRequest\x12\x10\n" +
+	"\x03sql\x18\x01 \x01(\tR\x03sql\x12\x1f\n" +
+	"\n" +
+	"schema_sql\x18\x02 \x01(\tH\x00R\tschemaSql\x12G\n" +
+	"\x11connection_params\x18\x03 \x01(\v2\x18.engine.ConnectionParamsH\x00R\x10connectionParamsB\x0f\n" +
+	"\rschema_source\"\x9d\x01\n" +
+	"\x10ConnectionParams\x12\x10\n" +
+	"\x03dsn\x18\x01 \x01(\tR\x03dsn\x12<\n" +
+	"\x06params\x18\x02 \x03(\v2$.engine.ConnectionParams.ParamsEntryR\x06params\x1a9\n" +
+	"\vParamsEntry\x12\x10\n" +
+	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
+	"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xad\x01\n" +
+	"\tStatement\x12\x12\n" +
+	"\x04name\x18\x01 \x01(\tR\x04name\x12\x1d\n" +
+	"\x03cmd\x18\x02 \x01(\x0e2\v.engine.CmdR\x03cmd\x12\x10\n" +
+	"\x03sql\x18\x03 \x01(\tR\x03sql\x121\n" +
+	"\n" +
+	"parameters\x18\x04 \x03(\v2\x11.engine.ParameterR\n" +
+	"parameters\x12(\n" +
+	"\acolumns\x18\x05 \x03(\v2\x0e.engine.ColumnR\acolumns\"m\n" +
+	"\rParseResponse\x121\n" +
+	"\n" +
+	"statements\x18\x01 \x03(\v2\x11.engine.StatementR\n" +
+	"statements\x12)\n" +
+	"\acatalog\x18\x02 \x01(\v2\x0f.engine.CatalogR\acatalog\"R\n" +
+	"\n" +
+	"Identifier\x12\x18\n" +
+	"\acatalog\x18\x01 \x01(\tR\acatalog\x12\x16\n" +
+	"\x06schema\x18\x02 \x01(\tR\x06schema\x12\x12\n" +
+	"\x04name\x18\x03 \x01(\tR\x04name\"\x8f\x01\n" +
+	"\aCatalog\x12\x18\n" +
+	"\acomment\x18\x01 \x01(\tR\acomment\x12%\n" +
+	"\x0edefault_schema\x18\x02 \x01(\tR\rdefaultSchema\x12\x12\n" +
+	"\x04name\x18\x03 \x01(\tR\x04name\x12/\n" +
+	"\aschemas\x18\x04 \x03(\v2\x15.engine.CatalogSchemaR\aschemas\"k\n" +
+	"\rCatalogSchema\x12\x18\n" +
+	"\acomment\x18\x01 \x01(\tR\acomment\x12\x12\n" +
+	"\x04name\x18\x02 \x01(\tR\x04name\x12,\n" +
+	"\x06tables\x18\x03 \x03(\v2\x14.engine.CatalogTableR\x06tables\"\x7f\n" +
+	"\fCatalogTable\x12$\n" +
+	"\x03rel\x18\x01 \x01(\v2\x12.engine.IdentifierR\x03rel\x12/\n" +
+	"\acolumns\x18\x02 \x03(\v2\x15.engine.CatalogColumnR\acolumns\x12\x18\n" +
+	"\acomment\x18\x03 \x01(\tR\acomment\"\xfc\x01\n" +
+	"\rCatalogColumn\x12\x12\n" +
+	"\x04name\x18\x01 \x01(\tR\x04name\x12\x19\n" +
+	"\bnot_null\x18\x03 \x01(\bR\anotNull\x12\x19\n" +
+	"\bis_array\x18\x04 \x01(\bR\aisArray\x12\x18\n" +
+	"\acomment\x18\x05 \x01(\tR\acomment\x12\x16\n" +
+	"\x06length\x18\x06 \x01(\x05R\x06length\x12\x1d\n" +
+	"\n" +
+	"array_dims\x18\x11 \x01(\x05R\tarrayDims\x12&\n" +
+	"\x04type\x18\f \x01(\v2\x12.engine.IdentifierR\x04type\x12(\n" +
+	"\x05table\x18\n" +
+	" \x01(\v2\x12.engine.IdentifierR\x05table\"\xae\x01\n" +
+	"\tParameter\x12\x12\n" +
+	"\x04name\x18\x01 \x01(\tR\x04name\x12\x1a\n" +
+	"\bposition\x18\x02 \x01(\x05R\bposition\x12\x1b\n" +
+	"\tdata_type\x18\x03 \x01(\tR\bdataType\x12\x1a\n" +
+	"\bnullable\x18\x04 \x01(\bR\bnullable\x12\x19\n" +
+	"\bis_array\x18\x05 \x01(\bR\aisArray\x12\x1d\n" +
+	"\n" +
+	"array_dims\x18\x06 \x01(\x05R\tarrayDims\"\xcf\x01\n" +
+	"\x06Column\x12\x12\n" +
+	"\x04name\x18\x01 \x01(\tR\x04name\x12\x1b\n" +
+	"\tdata_type\x18\x02 \x01(\tR\bdataType\x12\x1a\n" +
+	"\bnullable\x18\x03 \x01(\bR\bnullable\x12\x19\n" +
+	"\bis_array\x18\x04 \x01(\bR\aisArray\x12\x1d\n" +
+	"\n" +
+	"array_dims\x18\x05 \x01(\x05R\tarrayDims\x12\x1d\n" +
+	"\n" +
+	"table_name\x18\x06 \x01(\tR\ttableName\x12\x1f\n" +
+	"\vschema_name\x18\a \x01(\tR\n" +
+	"schemaName*\xcf\x01\n" +
+	"\x03Cmd\x12\x13\n" +
+	"\x0fCMD_UNSPECIFIED\x10\x00\x12\v\n" +
+	"\aCMD_ONE\x10\x01\x12\f\n" +
+	"\bCMD_MANY\x10\x02\x12\f\n" +
+	"\bCMD_EXEC\x10\x03\x12\x13\n" +
+	"\x0fCMD_EXEC_RESULT\x10\x04\x12\x11\n" +
+	"\rCMD_EXEC_ROWS\x10\x05\x12\x14\n" +
+	"\x10CMD_EXEC_LAST_ID\x10\x06\x12\x11\n" +
+	"\rCMD_COPY_FROM\x10\a\x12\x12\n" +
+	"\x0eCMD_BATCH_EXEC\x10\b\x12\x12\n" +
+	"\x0eCMD_BATCH_MANY\x10\t\x12\x11\n" +
+	"\rCMD_BATCH_ONE\x10\n" +
+	"2E\n" +
+	"\rEngineService\x124\n" +
+	"\x05Parse\x12\x14.engine.ParseRequest\x1a\x15.engine.ParseResponseb\x06proto3"
+
+var (
+	file_protos_engine_engine_proto_rawDescOnce sync.Once
+	file_protos_engine_engine_proto_rawDescData []byte
+)
+
+func file_protos_engine_engine_proto_rawDescGZIP() []byte {
+	file_protos_engine_engine_proto_rawDescOnce.Do(func() {
+		file_protos_engine_engine_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_protos_engine_engine_proto_rawDesc), len(file_protos_engine_engine_proto_rawDesc)))
+	})
+	return file_protos_engine_engine_proto_rawDescData
+}
+
+var file_protos_engine_engine_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_protos_engine_engine_proto_msgTypes = make([]protoimpl.MessageInfo, 12)
+var file_protos_engine_engine_proto_goTypes = []any{
+	(Cmd)(0),                 // 0: engine.Cmd
+	(*ParseRequest)(nil),     // 1: engine.ParseRequest
+	(*ConnectionParams)(nil), // 2: engine.ConnectionParams
+	(*Statement)(nil),        // 3: engine.Statement
+	(*ParseResponse)(nil),    // 4: engine.ParseResponse
+	(*Identifier)(nil),       // 5: engine.Identifier
+	(*Catalog)(nil),          // 6: engine.Catalog
+	(*CatalogSchema)(nil),    // 7: engine.CatalogSchema
+	(*CatalogTable)(nil),     // 8: engine.CatalogTable
+	(*CatalogColumn)(nil),    // 9: engine.CatalogColumn
+	(*Parameter)(nil),        // 10: engine.Parameter
+	(*Column)(nil),           // 11: engine.Column
+	nil,                      // 12: engine.ConnectionParams.ParamsEntry
+}
+var file_protos_engine_engine_proto_depIdxs = []int32{
+	2,  // 0: engine.ParseRequest.connection_params:type_name -> engine.ConnectionParams
+	12, // 1: engine.ConnectionParams.params:type_name -> engine.ConnectionParams.ParamsEntry
+	0,  // 2: engine.Statement.cmd:type_name -> engine.Cmd
+	10, // 3: engine.Statement.parameters:type_name -> engine.Parameter
+	11, // 4: engine.Statement.columns:type_name -> engine.Column
+	3,  // 5: engine.ParseResponse.statements:type_name -> engine.Statement
+	6,  // 6: engine.ParseResponse.catalog:type_name -> engine.Catalog
+	7,  // 7: engine.Catalog.schemas:type_name -> engine.CatalogSchema
+	8,  // 8: engine.CatalogSchema.tables:type_name -> engine.CatalogTable
+	5,  // 9: engine.CatalogTable.rel:type_name -> engine.Identifier
+	9,  // 10: engine.CatalogTable.columns:type_name -> engine.CatalogColumn
+	5,  // 11: engine.CatalogColumn.type:type_name -> engine.Identifier
+	5,  // 12: engine.CatalogColumn.table:type_name -> engine.Identifier
+	1,  // 13: engine.EngineService.Parse:input_type -> engine.ParseRequest
+	4,  // 14: engine.EngineService.Parse:output_type -> engine.ParseResponse
+	14, // [14:15] is the sub-list for method output_type
+	13, // [13:14] is the sub-list for method input_type
+	13, // [13:13] is the sub-list for extension type_name
+	13, // [13:13] is the sub-list for extension extendee
+	0,  // [0:13] is the sub-list for field type_name
+}
+
+func init() { file_protos_engine_engine_proto_init() }
+func file_protos_engine_engine_proto_init() {
+	if File_protos_engine_engine_proto != nil {
+		return
+	}
+	file_protos_engine_engine_proto_msgTypes[0].OneofWrappers = []any{
+		(*ParseRequest_SchemaSql)(nil),
+		(*ParseRequest_ConnectionParams)(nil),
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: unsafe.Slice(unsafe.StringData(file_protos_engine_engine_proto_rawDesc), len(file_protos_engine_engine_proto_rawDesc)),
+			NumEnums:      1,
+			NumMessages:   12,
+			NumExtensions: 0,
+			NumServices:   1,
+		},
+		GoTypes:           file_protos_engine_engine_proto_goTypes,
+		DependencyIndexes: file_protos_engine_engine_proto_depIdxs,
+		EnumInfos:         file_protos_engine_engine_proto_enumTypes,
+		MessageInfos:      file_protos_engine_engine_proto_msgTypes,
+	}.Build()
+	File_protos_engine_engine_proto = out.File
+	file_protos_engine_engine_proto_goTypes = nil
+	file_protos_engine_engine_proto_depIdxs = nil
+}
diff --git a/pkg/engine/engine_grpc.pb.go b/pkg/engine/engine_grpc.pb.go
new file mode 100644
index 0000000000..600130c84e
--- /dev/null
+++ b/pkg/engine/engine_grpc.pb.go
@@ -0,0 +1,133 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.6.0
+// - protoc             v6.32.1
+// source: protos/engine/engine.proto
+
+package engine
+
+import (
+	context "context"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.64.0 or later.
+const _ = grpc.SupportPackageIsVersion9
+
+const (
+	EngineService_Parse_FullMethodName = "/engine.EngineService/Parse"
+)
+
+// EngineServiceClient is the client API for EngineService service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+//
+// EngineService defines the interface for database engine plugins.
+// Engine plugins are responsible for parsing SQL statements and extracting
+// information about parameters and result columns.
+type EngineServiceClient interface {
+	// Parse parses SQL statements and returns the processed SQL along with
+	// information about parameters and result columns.
+	Parse(ctx context.Context, in *ParseRequest, opts ...grpc.CallOption) (*ParseResponse, error)
+}
+
+type engineServiceClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewEngineServiceClient(cc grpc.ClientConnInterface) EngineServiceClient {
+	return &engineServiceClient{cc}
+}
+
+func (c *engineServiceClient) Parse(ctx context.Context, in *ParseRequest, opts ...grpc.CallOption) (*ParseResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ParseResponse)
+	err := c.cc.Invoke(ctx, EngineService_Parse_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// EngineServiceServer is the server API for EngineService service.
+// All implementations must embed UnimplementedEngineServiceServer
+// for forward compatibility.
+//
+// EngineService defines the interface for database engine plugins.
+// Engine plugins are responsible for parsing SQL statements and extracting
+// information about parameters and result columns.
+type EngineServiceServer interface {
+	// Parse parses SQL statements and returns the processed SQL along with
+	// information about parameters and result columns.
+	Parse(context.Context, *ParseRequest) (*ParseResponse, error)
+	mustEmbedUnimplementedEngineServiceServer()
+}
+
+// UnimplementedEngineServiceServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedEngineServiceServer struct{}
+
+func (UnimplementedEngineServiceServer) Parse(context.Context, *ParseRequest) (*ParseResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method Parse not implemented")
+}
+func (UnimplementedEngineServiceServer) mustEmbedUnimplementedEngineServiceServer() {}
+func (UnimplementedEngineServiceServer) testEmbeddedByValue()                       {}
+
+// UnsafeEngineServiceServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to EngineServiceServer will
+// result in compilation errors.
+type UnsafeEngineServiceServer interface {
+	mustEmbedUnimplementedEngineServiceServer()
+}
+
+func RegisterEngineServiceServer(s grpc.ServiceRegistrar, srv EngineServiceServer) {
+	// If the following call panics, it indicates UnimplementedEngineServiceServer was
+	// embedded by pointer and is nil.  This will cause panics if an
+	// unimplemented method is ever invoked, so we test this at initialization
+	// time to prevent it from happening at runtime later due to I/O.
+	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+		t.testEmbeddedByValue()
+	}
+	s.RegisterService(&EngineService_ServiceDesc, srv)
+}
+
+func _EngineService_Parse_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ParseRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(EngineServiceServer).Parse(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: EngineService_Parse_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(EngineServiceServer).Parse(ctx, req.(*ParseRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// EngineService_ServiceDesc is the grpc.ServiceDesc for EngineService service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var EngineService_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "engine.EngineService",
+	HandlerType: (*EngineServiceServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "Parse",
+			Handler:    _EngineService_Parse_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "protos/engine/engine.proto",
+}
diff --git a/pkg/engine/helpers.go b/pkg/engine/helpers.go
new file mode 100644
index 0000000000..925aa94086
--- /dev/null
+++ b/pkg/engine/helpers.go
@@ -0,0 +1,136 @@
+// Package engine helpers for splitting query files and parsing " name: X :cmd" comments.
+// Plugins may use these to split query.sql into blocks and build Statement metadata.
+// ParseNameAndCmd and QueryBlocks delegate to internal/metadata so behavior matches
+// the built-in engines (postgresql, mysql, sqlite) exactly.
+
+package engine
+
+import (
+	"strings"
+
+	"github.com/sqlc-dev/sqlc/internal/metadata"
+	"github.com/sqlc-dev/sqlc/internal/source"
+)
+
+// CommentSyntax defines which comment styles to recognize (e.g. "--", "/* */", "#").
+// Same meaning as source.CommentSyntax / metadata.CommentSyntax.
+type CommentSyntax struct {
+	Dash      bool // "--"
+	SlashStar bool // "/* */"
+	Hash      bool // "#"
+}
+
+func toMetadataCommentSyntax(s CommentSyntax) metadata.CommentSyntax {
+	return metadata.CommentSyntax(source.CommentSyntax{
+		Dash: s.Dash, SlashStar: s.SlashStar, Hash: s.Hash,
+	})
+}
+
+// CmdToString returns the sqlc command string for c (e.g. ":one", ":many").
+// Returns "" for CMD_UNSPECIFIED.
+func CmdToString(c Cmd) string {
+	switch c {
+	case Cmd_CMD_ONE:
+		return ":one"
+	case Cmd_CMD_MANY:
+		return ":many"
+	case Cmd_CMD_EXEC:
+		return ":exec"
+	case Cmd_CMD_EXEC_RESULT:
+		return ":execresult"
+	case Cmd_CMD_EXEC_ROWS:
+		return ":execrows"
+	case Cmd_CMD_EXEC_LAST_ID:
+		return ":execlastid"
+	case Cmd_CMD_COPY_FROM:
+		return ":copyfrom"
+	case Cmd_CMD_BATCH_EXEC:
+		return ":batchexec"
+	case Cmd_CMD_BATCH_MANY:
+		return ":batchmany"
+	case Cmd_CMD_BATCH_ONE:
+		return ":batchone"
+	default:
+		return ""
+	}
+}
+
+// CmdFromString parses s (e.g. ":one", ":many") into Cmd. Returns (cmd, true) on success.
+func CmdFromString(s string) (Cmd, bool) {
+	switch strings.TrimSpace(s) {
+	case ":one":
+		return Cmd_CMD_ONE, true
+	case ":many":
+		return Cmd_CMD_MANY, true
+	case ":exec":
+		return Cmd_CMD_EXEC, true
+	case ":execresult":
+		return Cmd_CMD_EXEC_RESULT, true
+	case ":execrows":
+		return Cmd_CMD_EXEC_ROWS, true
+	case ":execlastid":
+		return Cmd_CMD_EXEC_LAST_ID, true
+	case ":copyfrom":
+		return Cmd_CMD_COPY_FROM, true
+	case ":batchexec":
+		return Cmd_CMD_BATCH_EXEC, true
+	case ":batchmany":
+		return Cmd_CMD_BATCH_MANY, true
+	case ":batchone":
+		return Cmd_CMD_BATCH_ONE, true
+	default:
+		return Cmd_CMD_UNSPECIFIED, false
+	}
+}
+
+// ParseNameAndCmd parses a single comment line like "-- name: ListAuthors :many" or
+// "# name: GetUser :one" or "/* name: CreateFoo :exec */". Returns (name, cmd, true) on success.
+// Uses metadata.ParseQueryNameAndType so behavior matches built-in engines exactly.
+func ParseNameAndCmd(line string, syntax CommentSyntax) (name string, cmd Cmd, ok bool) {
+	name, cmdStr, err := metadata.ParseQueryNameAndType(line+"\n", toMetadataCommentSyntax(syntax))
+	if err != nil || name == "" {
+		return "", Cmd_CMD_UNSPECIFIED, false
+	}
+	c, ok := CmdFromString(cmdStr)
+	if !ok {
+		return "", Cmd_CMD_UNSPECIFIED, false
+	}
+	return name, c, true
+}
+
+// QueryBlock is one named query block (from " name: X :cmd" to the next such line or EOF).
+type QueryBlock struct {
+	Name string
+	Cmd  Cmd
+	SQL  string
+}
+
+// QueryBlocks splits content into named query blocks. Each block runs from a
+// " name: X :cmd" line to the next such line (or EOF). Returns one entry per block.
+// Uses metadata.QueryBlocks so rules match the built-in engines exactly.
+func QueryBlocks(content string, syntax CommentSyntax) ([]QueryBlock, error) {
+	blocks, err := metadata.QueryBlocks(content, toMetadataCommentSyntax(syntax))
+	if err != nil {
+		return nil, err
+	}
+	out := make([]QueryBlock, 0, len(blocks))
+	for _, b := range blocks {
+		c, ok := CmdFromString(b.Cmd)
+		if !ok {
+			continue
+		}
+		out = append(out, QueryBlock{Name: b.Name, Cmd: c, SQL: b.SQL})
+	}
+	return out, nil
+}
+
+// StatementMeta builds a Statement with name, cmd, and sql set. Parameters and columns
+// are left empty for the plugin to fill. Plugins can use this after splitting query.sql
+// with QueryBlocks and then attach their own parameters/columns per block.
+func StatementMeta(name string, cmd Cmd, sql string) *Statement {
+	return &Statement{
+		Name: name,
+		Cmd:  cmd,
+		Sql:  sql,
+	}
+}
diff --git a/pkg/engine/helpers_test.go b/pkg/engine/helpers_test.go
new file mode 100644
index 0000000000..05cb767697
--- /dev/null
+++ b/pkg/engine/helpers_test.go
@@ -0,0 +1,141 @@
+package engine
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestCmdToString(t *testing.T) {
+	tests := []struct {
+		c    Cmd
+		want string
+	}{
+		{Cmd_CMD_UNSPECIFIED, ""},
+		{Cmd_CMD_ONE, ":one"},
+		{Cmd_CMD_MANY, ":many"},
+		{Cmd_CMD_EXEC, ":exec"},
+		{Cmd_CMD_EXEC_RESULT, ":execresult"},
+		{Cmd_CMD_EXEC_ROWS, ":execrows"},
+		{Cmd_CMD_EXEC_LAST_ID, ":execlastid"},
+		{Cmd_CMD_COPY_FROM, ":copyfrom"},
+		{Cmd_CMD_BATCH_EXEC, ":batchexec"},
+		{Cmd_CMD_BATCH_MANY, ":batchmany"},
+		{Cmd_CMD_BATCH_ONE, ":batchone"},
+	}
+	for _, tt := range tests {
+		if got := CmdToString(tt.c); got != tt.want {
+			t.Errorf("CmdToString(%v) = %q, want %q", tt.c, got, tt.want)
+		}
+	}
+}
+
+func TestCmdFromString(t *testing.T) {
+	valid := []string{":one", ":many", ":exec", ":execresult", ":execrows", ":execlastid", ":copyfrom", ":batchexec", ":batchmany", ":batchone"}
+	for _, s := range valid {
+		c, ok := CmdFromString(s)
+		if !ok {
+			t.Errorf("CmdFromString(%q) ok=false", s)
+			continue
+		}
+		if CmdToString(c) != s {
+			t.Errorf("CmdFromString(%q) -> %v -> CmdToString = %q", s, c, CmdToString(c))
+		}
+	}
+	_, ok := CmdFromString(":invalid")
+	if ok {
+		t.Error("CmdFromString(:invalid) ok=true")
+	}
+}
+
+func TestParseNameAndCmd(t *testing.T) {
+	dash := CommentSyntax{Dash: true}
+	hash := CommentSyntax{Hash: true}
+	slash := CommentSyntax{SlashStar: true}
+
+	t.Run("Parse", func(t *testing.T) {
+		for _, line := range []string{
+			`-- name: CreateFoo, :one`,
+			`-- name: 9Foo :one`,
+			`-- name: CreateFoo :two`,
+			`-- name: CreateFoo`,
+			`-- name: CreateFoo :one something`,
+			`-- name: `,
+			`--name: CreateFoo :one`,
+			`--name CreateFoo :one`,
+		} {
+			t.Run(line, func(t *testing.T) {
+				if _, _, ok := ParseNameAndCmd(line, dash); ok {
+					t.Errorf("expected invalid: %q", line)
+				}
+			})
+		}
+	})
+
+	t.Run("CommentSyntax", func(t *testing.T) {
+		for line, syn := range map[string]CommentSyntax{
+			`-- name: CreateFoo :one`:    dash,
+			`# name: CreateFoo :one`:     hash,
+			`/* name: CreateFoo :one */`: slash,
+		} {
+			t.Run(line, func(t *testing.T) {
+				name, cmd, ok := ParseNameAndCmd(line, syn)
+				if !ok {
+					t.Errorf("expected valid %q", line)
+					return
+				}
+				if name != "CreateFoo" {
+					t.Errorf("ParseNameAndCmd(%q) name = %q", line, name)
+				}
+				if cmd != Cmd_CMD_ONE {
+					t.Errorf("ParseNameAndCmd(%q) cmd = %v", line, cmd)
+				}
+			})
+		}
+	})
+
+	t.Run("Many", func(t *testing.T) {
+		name, cmd, ok := ParseNameAndCmd("-- name: ListAuthors :many", dash)
+		if !ok || name != "ListAuthors" || cmd != Cmd_CMD_MANY {
+			t.Errorf("-- name: ListAuthors :many -> name=%q cmd=%v ok=%v", name, cmd, ok)
+		}
+	})
+}
+
+func TestQueryBlocks(t *testing.T) {
+	syntax := CommentSyntax{Dash: true, SlashStar: true}
+	content := `-- name: GetUser :one
+SELECT id, name FROM users WHERE id = $1
+
+-- name: ListUsers :many
+SELECT id, name FROM users ORDER BY id
+`
+	blocks, err := QueryBlocks(content, syntax)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(blocks) != 2 {
+		t.Fatalf("len(blocks)=%d, want 2", len(blocks))
+	}
+	if blocks[0].Name != "GetUser" || blocks[0].Cmd != Cmd_CMD_ONE {
+		t.Errorf("block0: Name=%q Cmd=%v", blocks[0].Name, blocks[0].Cmd)
+	}
+	if blocks[1].Name != "ListUsers" || blocks[1].Cmd != Cmd_CMD_MANY {
+		t.Errorf("block1: Name=%q Cmd=%v", blocks[1].Name, blocks[1].Cmd)
+	}
+	if !strings.Contains(blocks[0].SQL, "SELECT id, name FROM users WHERE id = $1") {
+		t.Errorf("block0 SQL missing expected fragment: %s", blocks[0].SQL)
+	}
+	if !strings.Contains(blocks[1].SQL, "SELECT id, name FROM users ORDER BY id") {
+		t.Errorf("block1 SQL missing expected fragment: %s", blocks[1].SQL)
+	}
+}
+
+func TestStatementMeta(t *testing.T) {
+	st := StatementMeta("GetUser", Cmd_CMD_ONE, "SELECT 1")
+	if st.Name != "GetUser" || st.Cmd != Cmd_CMD_ONE || st.Sql != "SELECT 1" {
+		t.Errorf("StatementMeta: Name=%q Cmd=%v Sql=%q", st.Name, st.Cmd, st.Sql)
+	}
+	if st.Parameters != nil || st.Columns != nil {
+		t.Error("StatementMeta should leave Parameters/Columns nil")
+	}
+}
diff --git a/pkg/engine/sdk.go b/pkg/engine/sdk.go
new file mode 100644
index 0000000000..32e7c2ff81
--- /dev/null
+++ b/pkg/engine/sdk.go
@@ -0,0 +1,110 @@
+// Package engine provides types and utilities for building sqlc database engine plugins.
+//
+// Engine plugins allow external database backends to be used with sqlc.
+// Plugins communicate with sqlc via Protocol Buffers over stdin/stdout.
+//
+// # Compatibility
+//
+// Go plugins that import this package are guaranteed to be compatible with sqlc
+// at compile time. If the types change incompatibly, the plugin simply won't
+// compile until it's updated to match the new interface.
+//
+// The Protocol Buffer schema is published at buf.build/sqlc/sqlc and ensures
+// binary compatibility between sqlc and plugins.
+//
+// # Generating engine.pb.go and engine_grpc.pb.go
+//
+// Run from the repository root:
+//
+//	make proto-engine-plugin
+//
+// or:
+//
+//	protoc -I. \
+//		--go_out=. --go_opt=module=github.com/sqlc-dev/sqlc --go_opt=Mprotos/engine/engine.proto=github.com/sqlc-dev/sqlc/pkg/engine \
+//		--go-grpc_out=. --go-grpc_opt=module=github.com/sqlc-dev/sqlc --go-grpc_opt=Mprotos/engine/engine.proto=github.com/sqlc-dev/sqlc/pkg/engine \
+//		protos/engine/engine.proto
+//
+// Example plugin:
+//
+//	package main
+//
+//	import "github.com/sqlc-dev/sqlc/pkg/engine"
+//
+//	func main() {
+//		engine.Run(engine.Handler{
+//			PluginName:    "my-plugin",
+//			PluginVersion: "1.0.0",
+//			Parse:         handleParse,
+//		})
+//	}
+//
+//go:generate sh -c "cd ../.. && make proto-engine-plugin"
+package engine
+
+import (
+	"fmt"
+	"io"
+	"os"
+
+	"google.golang.org/protobuf/proto"
+)
+
+// Handler contains the functions that implement the engine plugin interface.
+// All types used are Protocol Buffer messages defined in engine.proto.
+type Handler struct {
+	PluginName    string
+	PluginVersion string
+
+	Parse func(*ParseRequest) (*ParseResponse, error)
+}
+
+// Run runs the engine plugin with the given handler.
+// It reads a protobuf request from stdin and writes a protobuf response to stdout.
+func Run(h Handler) {
+	if err := run(h, os.Args, os.Stdin, os.Stdout, os.Stderr); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+}
+
+func run(h Handler, args []string, stdin io.Reader, stdout, stderr io.Writer) error {
+	if len(args) < 2 {
+		return fmt.Errorf("usage: %s <method>", args[0])
+	}
+
+	method := args[1]
+	input, err := io.ReadAll(stdin)
+	if err != nil {
+		return fmt.Errorf("reading stdin: %w", err)
+	}
+
+	var output proto.Message
+
+	switch method {
+	case "parse":
+		var req ParseRequest
+		if err := proto.Unmarshal(input, &req); err != nil {
+			return fmt.Errorf("parsing request: %w", err)
+		}
+		if h.Parse == nil {
+			return fmt.Errorf("parse not implemented")
+		}
+		output, err = h.Parse(&req)
+
+	default:
+		return fmt.Errorf("unknown method: %s", method)
+	}
+
+	if err != nil {
+		return err
+	}
+
+	data, err := proto.Marshal(output)
+	if err != nil {
+		return fmt.Errorf("marshaling response: %w", err)
+	}
+
+	_, err = stdout.Write(data)
+	return err
+}
diff --git a/protos/engine/engine.proto b/protos/engine/engine.proto
new file mode 100644
index 0000000000..3a612081dd
--- /dev/null
+++ b/protos/engine/engine.proto
@@ -0,0 +1,158 @@
+syntax = "proto3";
+
+package engine;
+
+// EngineService defines the interface for database engine plugins.
+// Engine plugins are responsible for parsing SQL statements and extracting
+// information about parameters and result columns.
+service EngineService {
+  // Parse parses SQL statements and returns the processed SQL along with
+  // information about parameters and result columns.
+  rpc Parse (ParseRequest) returns (ParseResponse);
+}
+
+// ParseRequest contains the SQL to parse and either schema or connection parameters.
+message ParseRequest {
+  // The SQL query text to parse.
+  string sql = 1;
+  
+  // Either schema SQL or connection parameters for database-only mode.
+  oneof schema_source {
+    // Schema SQL text (schema.sql) for schema-based parsing.
+    string schema_sql = 2;
+    
+    // Connection parameters for database-only mode.
+    ConnectionParams connection_params = 3;
+  }
+}
+
+// ConnectionParams contains database connection parameters for database-only mode.
+message ConnectionParams {
+  // Database connection string or DSN.
+  string dsn = 1;
+  
+  // Additional connection parameters as key-value pairs.
+  map<string, string> params = 2;
+}
+
+// Cmd is the query command/type. Matches sqlc's ":one", ":many", ":exec", etc.
+enum Cmd {
+  CMD_UNSPECIFIED = 0;
+  CMD_ONE = 1;
+  CMD_MANY = 2;
+  CMD_EXEC = 3;
+  CMD_EXEC_RESULT = 4;
+  CMD_EXEC_ROWS = 5;
+  CMD_EXEC_LAST_ID = 6;
+  CMD_COPY_FROM = 7;
+  CMD_BATCH_EXEC = 8;
+  CMD_BATCH_MANY = 9;
+  CMD_BATCH_ONE = 10;
+}
+
+// Statement is one parsed query block: name/cmd from " name: X :cmd", plus sql and type info.
+message Statement {
+  // Query name (from "-- name: GetUser" etc.).
+  string name = 1;
+  // Command/type for this statement.
+  Cmd cmd = 2;
+  // Processed SQL for this statement (wildcards expanded if applicable).
+  string sql = 3;
+  // Parameters for this statement.
+  repeated Parameter parameters = 4;
+  // Result columns for this statement.
+  repeated Column columns = 5;
+}
+
+// ParseResponse contains the processed statements and optional catalog. The plugin receives
+// the full query file and schema (or connection params); it returns one Statement per query
+// block. Catalog is derived from schema_sql or connection_params and is passed to codegen.
+message ParseResponse {
+  repeated Statement statements = 1;
+  Catalog catalog = 2;
+}
+
+// Identifier identifies a catalog/schema/name (used in Catalog types).
+message Identifier {
+  string catalog = 1;
+  string schema = 2;
+  string name = 3;
+}
+
+// Catalog mirrors plugin.Catalog so engine plugins can return schema without importing plugin.
+// The sqlc app converts it to plugin.Catalog when calling codegen.
+message Catalog {
+  string comment = 1;
+  string default_schema = 2;
+  string name = 3;
+  repeated CatalogSchema schemas = 4;
+}
+
+message CatalogSchema {
+  string comment = 1;
+  string name = 2;
+  repeated CatalogTable tables = 3;
+}
+
+message CatalogTable {
+  Identifier rel = 1;
+  repeated CatalogColumn columns = 2;
+  string comment = 3;
+}
+
+message CatalogColumn {
+  string name = 1;
+  bool not_null = 3;
+  bool is_array = 4;
+  string comment = 5;
+  int32 length = 6;
+  int32 array_dims = 17;
+  Identifier type = 12;
+  Identifier table = 10;
+}
+
+// Parameter represents a query parameter.
+message Parameter {
+  // Parameter name (if named) or empty for positional parameters.
+  string name = 1;
+  
+  // Parameter position (1-based) for positional parameters.
+  int32 position = 2;
+  
+  // SQL data type of the parameter.
+  string data_type = 3;
+  
+  // Whether the parameter is nullable.
+  bool nullable = 4;
+  
+  // Whether the parameter is an array type.
+  bool is_array = 5;
+  
+  // Array dimensions if is_array is true.
+  int32 array_dims = 6;
+}
+
+// Column represents a result column.
+message Column {
+  // Column name.
+  string name = 1;
+  
+  // SQL data type of the column.
+  string data_type = 2;
+  
+  // Whether the column is nullable.
+  bool nullable = 3;
+  
+  // Whether the column is an array type.
+  bool is_array = 4;
+  
+  // Array dimensions if is_array is true.
+  int32 array_dims = 5;
+  
+  // Table name this column belongs to (if known).
+  string table_name = 6;
+  
+  // Schema name this column belongs to (if known).
+  string schema_name = 7;
+}
+
