From 88f837a33c9e187857ff4d81112f32df6767110f Mon Sep 17 00:00:00 2001 From: Chenghan Ying <125171961+roychying@users.noreply.github.com> Date: Mon, 21 Sep 2026 11:13:31 -0700 Subject: [PATCH] refactor(storage): group MySQL schema into readmodel and pipeline packages by owning service --- Makefile | 2 +- .../storage/mysql/schema/BUILD.bazel | 8 ++++- .../extension/storage/mysql/schema/README.md | 9 ++++++ .../storage/mysql/schema/pipeline/BUILD.bazel | 5 ++++ .../mysql/schema/{ => pipeline}/batch.sql | 0 .../schema/{ => pipeline}/batch_dependent.sql | 0 .../mysql/schema/{ => pipeline}/build.sql | 0 .../mysql/schema/{ => pipeline}/change.sql | 0 .../schema/{ => pipeline}/path_build.sql | 0 .../{ => pipeline}/queue_batch_state.sql | 0 .../mysql/schema/{ => pipeline}/request.sql | 0 .../schema/{ => pipeline}/request_batch.sql | 0 .../{ => pipeline}/speculation_path_set.sql | 0 .../mysql/schema/readmodel/BUILD.bazel | 5 ++++ .../change_uri_request_mapping.sql | 0 .../schema/{ => readmodel}/request_log.sql | 0 .../{ => readmodel}/request_summary.sql | 0 .../request_summary_by_queue.sql | 0 test/testutil/schema.go | 26 ++++++++++++---- tool/linter/queueshard/main.go | 30 +++++++++++++++++-- 20 files changed, 75 insertions(+), 10 deletions(-) create mode 100644 submitqueue/extension/storage/mysql/schema/pipeline/BUILD.bazel rename submitqueue/extension/storage/mysql/schema/{ => pipeline}/batch.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => pipeline}/batch_dependent.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => pipeline}/build.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => pipeline}/change.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => pipeline}/path_build.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => pipeline}/queue_batch_state.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => pipeline}/request.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => pipeline}/request_batch.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => pipeline}/speculation_path_set.sql (100%) create mode 100644 submitqueue/extension/storage/mysql/schema/readmodel/BUILD.bazel rename submitqueue/extension/storage/mysql/schema/{ => readmodel}/change_uri_request_mapping.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => readmodel}/request_log.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => readmodel}/request_summary.sql (100%) rename submitqueue/extension/storage/mysql/schema/{ => readmodel}/request_summary_by_queue.sql (100%) diff --git a/Makefile b/Makefile index b15be5161..12d9f90cc 100644 --- a/Makefile +++ b/Makefile @@ -386,7 +386,7 @@ local-submitqueue-gateway-stop: ## Stop Gateway service local-init-submitqueue-schemas: ## Manually apply all database schemas @echo "Applying storage schema to mysql-app..." - @for file in submitqueue/extension/storage/mysql/schema/*.sql; do \ + @for file in submitqueue/extension/storage/mysql/schema/*/*.sql; do \ echo " - Applying $$(basename $$file)..."; \ docker exec -i $(SUBMITQUEUE_LOCAL_PROJECT)-mysql-app-1 mysql -uroot -proot submitqueue < $$file 2>&1 | grep -v "Using a password" || true; \ done diff --git a/submitqueue/extension/storage/mysql/schema/BUILD.bazel b/submitqueue/extension/storage/mysql/schema/BUILD.bazel index 3412d7733..489f5b9af 100644 --- a/submitqueue/extension/storage/mysql/schema/BUILD.bazel +++ b/submitqueue/extension/storage/mysql/schema/BUILD.bazel @@ -1,5 +1,11 @@ +# The whole schema. A deployment that colocates gateway and orchestrator on one +# database uses this alone; the per-owner packages exist for deployments that +# give each service its own database, and are not a statement that they must. filegroup( name = "schema", - srcs = glob(["*.sql"]), + srcs = [ + "//submitqueue/extension/storage/mysql/schema/pipeline", + "//submitqueue/extension/storage/mysql/schema/readmodel", + ], visibility = ["//visibility:public"], ) diff --git a/submitqueue/extension/storage/mysql/schema/README.md b/submitqueue/extension/storage/mysql/schema/README.md index 59fef6c4b..4f823701e 100644 --- a/submitqueue/extension/storage/mysql/schema/README.md +++ b/submitqueue/extension/storage/mysql/schema/README.md @@ -1,5 +1,14 @@ # MySQL Schema +## Layout + +Tables are grouped by the service that owns them, one Bazel package each: + +- `readmodel/` — the gateway-owned read model: the append-only request log and the three materialized projections behind request-summary retrieval and `List` (see [Gateway request read model](#gateway-request-read-model) below and [doc/rfc/submitqueue/status-list-api.md](../../../../../doc/rfc/submitqueue/status-list-api.md)). +- `pipeline/` — orchestrator pipeline working state: requests, batches, builds, changes, and speculation. Different retention semantics from the read model, and never read by the gateway APIs. + +The parent `:schema` filegroup is the union of both and remains the default: a deployment that colocates gateway and orchestrator on one database depends on it alone. The per-owner packages exist so a deployment that gives each service its own database can provision exactly that service's tables; they are not a statement that it must. Adding a table means placing its `.sql` in the owning package — the grouping is the file's location, so there is no list to keep in sync. + ## Queue-leading primary keys Every table leads its primary key with `queue`: `request` and `batch` on `(queue, id)`, `build` on `(queue, id)`, `batch_dependent` on `(queue, batch_id)`, `request_batch` on `(queue, request_id, batch_id)`, `change` on `(queue, uri, request_id)`, `queue_batch_state` on `(queue, state, batch_id)`, `speculation_path_set` on `(queue, head)`, `request_summary` on `(queue, request_id)`, `request_log` on `(queue, request_id, timestamp_ms, salt)`, `change_uri_request_mapping` on `(queue, change_uri, received_at_ms, request_id)`, and `request_summary_by_queue` on `(queue, received_at_ms, request_id)`. A queue-bound store instance prefixes every read and stamps every write with its bound queue, so one queue's rows are unreachable through another queue's binding and every table is shardable by queue. `//tool/linter/queueshard` enforces this, and also rejects any secondary index that does not itself lead with `queue`, since such an index would reintroduce a cross-queue access path. diff --git a/submitqueue/extension/storage/mysql/schema/pipeline/BUILD.bazel b/submitqueue/extension/storage/mysql/schema/pipeline/BUILD.bazel new file mode 100644 index 000000000..44f7238c0 --- /dev/null +++ b/submitqueue/extension/storage/mysql/schema/pipeline/BUILD.bazel @@ -0,0 +1,5 @@ +filegroup( + name = "pipeline", + srcs = glob(["*.sql"]), + visibility = ["//visibility:public"], +) diff --git a/submitqueue/extension/storage/mysql/schema/batch.sql b/submitqueue/extension/storage/mysql/schema/pipeline/batch.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/batch.sql rename to submitqueue/extension/storage/mysql/schema/pipeline/batch.sql diff --git a/submitqueue/extension/storage/mysql/schema/batch_dependent.sql b/submitqueue/extension/storage/mysql/schema/pipeline/batch_dependent.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/batch_dependent.sql rename to submitqueue/extension/storage/mysql/schema/pipeline/batch_dependent.sql diff --git a/submitqueue/extension/storage/mysql/schema/build.sql b/submitqueue/extension/storage/mysql/schema/pipeline/build.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/build.sql rename to submitqueue/extension/storage/mysql/schema/pipeline/build.sql diff --git a/submitqueue/extension/storage/mysql/schema/change.sql b/submitqueue/extension/storage/mysql/schema/pipeline/change.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/change.sql rename to submitqueue/extension/storage/mysql/schema/pipeline/change.sql diff --git a/submitqueue/extension/storage/mysql/schema/path_build.sql b/submitqueue/extension/storage/mysql/schema/pipeline/path_build.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/path_build.sql rename to submitqueue/extension/storage/mysql/schema/pipeline/path_build.sql diff --git a/submitqueue/extension/storage/mysql/schema/queue_batch_state.sql b/submitqueue/extension/storage/mysql/schema/pipeline/queue_batch_state.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/queue_batch_state.sql rename to submitqueue/extension/storage/mysql/schema/pipeline/queue_batch_state.sql diff --git a/submitqueue/extension/storage/mysql/schema/request.sql b/submitqueue/extension/storage/mysql/schema/pipeline/request.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/request.sql rename to submitqueue/extension/storage/mysql/schema/pipeline/request.sql diff --git a/submitqueue/extension/storage/mysql/schema/request_batch.sql b/submitqueue/extension/storage/mysql/schema/pipeline/request_batch.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/request_batch.sql rename to submitqueue/extension/storage/mysql/schema/pipeline/request_batch.sql diff --git a/submitqueue/extension/storage/mysql/schema/speculation_path_set.sql b/submitqueue/extension/storage/mysql/schema/pipeline/speculation_path_set.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/speculation_path_set.sql rename to submitqueue/extension/storage/mysql/schema/pipeline/speculation_path_set.sql diff --git a/submitqueue/extension/storage/mysql/schema/readmodel/BUILD.bazel b/submitqueue/extension/storage/mysql/schema/readmodel/BUILD.bazel new file mode 100644 index 000000000..6bf9035dd --- /dev/null +++ b/submitqueue/extension/storage/mysql/schema/readmodel/BUILD.bazel @@ -0,0 +1,5 @@ +filegroup( + name = "readmodel", + srcs = glob(["*.sql"]), + visibility = ["//visibility:public"], +) diff --git a/submitqueue/extension/storage/mysql/schema/change_uri_request_mapping.sql b/submitqueue/extension/storage/mysql/schema/readmodel/change_uri_request_mapping.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/change_uri_request_mapping.sql rename to submitqueue/extension/storage/mysql/schema/readmodel/change_uri_request_mapping.sql diff --git a/submitqueue/extension/storage/mysql/schema/request_log.sql b/submitqueue/extension/storage/mysql/schema/readmodel/request_log.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/request_log.sql rename to submitqueue/extension/storage/mysql/schema/readmodel/request_log.sql diff --git a/submitqueue/extension/storage/mysql/schema/request_summary.sql b/submitqueue/extension/storage/mysql/schema/readmodel/request_summary.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/request_summary.sql rename to submitqueue/extension/storage/mysql/schema/readmodel/request_summary.sql diff --git a/submitqueue/extension/storage/mysql/schema/request_summary_by_queue.sql b/submitqueue/extension/storage/mysql/schema/readmodel/request_summary_by_queue.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/request_summary_by_queue.sql rename to submitqueue/extension/storage/mysql/schema/readmodel/request_summary_by_queue.sql diff --git a/test/testutil/schema.go b/test/testutil/schema.go index c736c88ad..e992bd64d 100644 --- a/test/testutil/schema.go +++ b/test/testutil/schema.go @@ -17,9 +17,11 @@ package testutil import ( "context" "database/sql" + "io/fs" "os" "path/filepath" "sort" + "strings" "testing" _ "github.com/go-sql-driver/mysql" @@ -46,19 +48,33 @@ func SchemaDir(relativePath string) string { return Runfile(relativePath) } -// ApplySchema reads all .sql files from the schema directory and executes them on the database. +// ApplySchema reads every .sql file under the schema directory, including +// subdirectories, and executes them on the database. A schema may group its +// tables into per-owner subpackages (see +// submitqueue/extension/storage/mysql/schema), so passing the root applies the +// whole schema regardless of how it is subdivided. func ApplySchema(t *testing.T, log *TestLogger, db *sql.DB, schemaDirectory string) { t.Helper() - files, err := filepath.Glob(filepath.Join(schemaDirectory, "*.sql")) - require.NoError(t, err, "failed to glob schema files") - require.NotEmpty(t, files, "no .sql schema files found in %s", schemaDirectory) + var files []string + err := filepath.WalkDir(schemaDirectory, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if !d.IsDir() && strings.HasSuffix(d.Name(), ".sql") { + files = append(files, path) + } + return nil + }) + require.NoError(t, err, "failed to walk schema files") + require.NotEmpty(t, files, "no .sql schema files found under %s", schemaDirectory) // Sort files to ensure deterministic schema application order. sort.Strings(files) for _, f := range files { - name := filepath.Base(f) + name, relErr := filepath.Rel(schemaDirectory, f) + require.NoError(t, relErr, "failed to relativize schema file %s", f) log.Logf("Applying schema: %s", name) content, err := os.ReadFile(f) diff --git a/tool/linter/queueshard/main.go b/tool/linter/queueshard/main.go index d900bc11d..ce8717b18 100644 --- a/tool/linter/queueshard/main.go +++ b/tool/linter/queueshard/main.go @@ -26,9 +26,11 @@ package main import ( "flag" "fmt" + "io/fs" "os" "path/filepath" "regexp" + "sort" "strings" ) @@ -40,7 +42,8 @@ var schemaShardColumns = map[string]map[string]bool{ "platform/extension/messagequeue/mysql/schema": {"tenant": true}, } -// schemaRoots are the directories scanned for table definitions. +// schemaRoots are the directory trees scanned for table definitions. A root's +// tables may be grouped into per-owner subpackages; the whole tree is scanned. var schemaRoots = []string{ "submitqueue/extension/storage/mysql/schema", "stovepipe/extension/storage/mysql/schema", @@ -74,9 +77,9 @@ func main() { var checked int for _, schemaRoot := range schemaRoots { shardColumns := schemaShardColumns[schemaRoot] - files, err := filepath.Glob(filepath.Join(root, schemaRoot, "*.sql")) + files, err := findSchemaFiles(filepath.Join(root, schemaRoot)) if err != nil { - fmt.Fprintf(os.Stderr, "error globbing %s: %v\n", schemaRoot, err) + fmt.Fprintf(os.Stderr, "error walking %s: %v\n", schemaRoot, err) os.Exit(1) } if len(files) == 0 { @@ -112,6 +115,27 @@ func main() { fmt.Printf("All %d tables are shardable.\n", checked) } +// findSchemaFiles returns every .sql file under dir, including subdirectories. +// A schema root may group its tables into per-owner subpackages, so the whole +// tree is scanned rather than only the root's own files. +func findSchemaFiles(dir string) ([]string, error) { + var files []string + err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if !d.IsDir() && strings.HasSuffix(d.Name(), ".sql") { + files = append(files, path) + } + return nil + }) + if err != nil { + return nil, err + } + sort.Strings(files) + return files, nil +} + // check returns the number of tables found in content and any violations. func check(file, content string, shardColumns map[string]bool) (int, []violation) { var violations []violation