From 90a1fc24e80465dfceb283586fb9851286f1e06a Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 08:26:27 -0500 Subject: [PATCH 01/12] feat(proto): add MysqlConnection to WarehouseConfig union Co-Authored-By: Claude Opus 4.8 --- protos/configs.proto | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/protos/configs.proto b/protos/configs.proto index a87ea207..32277ade 100644 --- a/protos/configs.proto +++ b/protos/configs.proto @@ -1028,6 +1028,17 @@ message SupabaseConnection { string connection_string = 4; } +message MysqlConnection { + string host = 1; + uint32 port = 2; + string database = 3; + string user = 4; + string password = 5; + + // SSL mode: "disable" | "require". MySQL/MariaDB over TLS. + string ssl_mode = 6; +} + // WarehouseConfig — discriminated union over connection variants. The // `kind:` YAML tag selects which `oneof` arm is unmarshalled. // @@ -1046,5 +1057,6 @@ message WarehouseConfig { BigQueryConnection bigquery = 1; PostgresConnection postgres = 2; SupabaseConnection supabase = 3; + MysqlConnection mysql = 4; } } From 56bc0f1868ea28de60818bd8ebc35e06a25fc01f Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 08:29:08 -0500 Subject: [PATCH 02/12] feat(core): accept mysql warehouse + MySQL backtick quoting - supportedWarehouses now includes mysql; main_test rejection case uses snowflake - CompilationSql resolveTarget emits `schema`.`name` for mysql (no catalog level) - indexAssertion backtick-quotes columns; sqlString uses backslash escaping for mysql - unit tests for resolveTarget/sqlString/indexAssertion mysql branches Co-Authored-By: Claude Opus 4.8 --- core/compilation_sql/compilation_sql_test.ts | 31 +++++++++++++++++++- core/compilation_sql/index.ts | 13 +++++++- core/main_test.ts | 6 ++-- core/workflow_settings.ts | 2 +- 4 files changed, 46 insertions(+), 6 deletions(-) diff --git a/core/compilation_sql/compilation_sql_test.ts b/core/compilation_sql/compilation_sql_test.ts index e2d12315..d478305e 100644 --- a/core/compilation_sql/compilation_sql_test.ts +++ b/core/compilation_sql/compilation_sql_test.ts @@ -45,10 +45,25 @@ suite("CompilationSql", () => { defaultSchema: "public" }); const compiler = new CompilationSql(config, "3.0.0"); - + expect(compiler.resolveTarget({ schema: "public", name: "my_table" })) .to.equal('"public"."my_table"'); }); + + test("MySQL: should format with backticks as `schema`.`name`", () => { + const config = sqlanvil.ProjectConfig.create({ + warehouse: "mysql", + defaultSchema: "my_db" + }); + const compiler = new CompilationSql(config, "3.0.0"); + + expect(compiler.resolveTarget({ schema: "my_db", name: "my_table" })) + .to.equal("`my_db`.`my_table`"); + + // MySQL has no catalog level, so any database is ignored. + expect(compiler.resolveTarget({ database: "ignored", schema: "my_db", name: "my_table" })) + .to.equal("`my_db`.`my_table`"); + }); }); suite("sqlString", () => { @@ -63,6 +78,12 @@ suite("CompilationSql", () => { const compiler = new CompilationSql(config, "3.0.0"); expect(compiler.sqlString("it's a \\test")).to.equal("'it''s a \\test'"); }); + + test("MySQL: escapes using backslashes", () => { + const config = sqlanvil.ProjectConfig.create({ warehouse: "mysql" }); + const compiler = new CompilationSql(config, "3.0.0"); + expect(compiler.sqlString("it's a \\test")).to.equal("'it\\'s a \\\\test'"); + }); }); suite("indexAssertion", () => { @@ -81,5 +102,13 @@ suite("CompilationSql", () => { expect(result).to.contain('"col1", "col2"'); expect(result).to.contain('FROM "my_schema"."my_table"'); }); + + test("MySQL: columns are backtick-quoted", () => { + const config = sqlanvil.ProjectConfig.create({ warehouse: "mysql" }); + const compiler = new CompilationSql(config, "3.0.0"); + const result = compiler.indexAssertion("`my_db`.`my_table`", ["col1", "col2"]); + expect(result).to.contain("`col1`, `col2`"); + expect(result).to.contain("FROM `my_db`.`my_table`"); + }); }); }); diff --git a/core/compilation_sql/index.ts b/core/compilation_sql/index.ts index 7598ab10..28a36bed 100644 --- a/core/compilation_sql/index.ts +++ b/core/compilation_sql/index.ts @@ -23,6 +23,12 @@ export class CompilationSql { return `"${database}"."${schema}"."${name}"`; } + if (this.warehouse === "mysql") { + // MySQL/MariaDB backtick dialect. MySQL has no catalog level, so the schema + // is the database: `schema`.`name`. Any `database` is ignored. + return `\`${schema}\`.\`${name}\``; + } + // Default to BigQuery backtick dialect: `database.schema.name` if (!database) { return `\`${schema}.${name}\``; @@ -35,7 +41,8 @@ export class CompilationSql { // Postgres/ANSI SQL standard single quote escaping (doubling up single quotes) return `'${stringContents.replace(/'/g, "''")}'`; } - // BigQuery backslash-based single quote escaping + // BigQuery and MySQL/MariaDB both treat backslash as an escape char by default, + // so escape backslashes then single quotes. return `'${stringContents.replace(/\\/g, "\\\\").replace(/'/g, "\\'")}'`; } @@ -45,6 +52,10 @@ export class CompilationSql { // Double quote columns to handle case sensitivity and reserved SQL keywords return `"${col.replace(/"/g, '""')}"`; } + if (this.warehouse === "mysql") { + // MySQL/MariaDB backtick-quote columns; escape embedded backticks by doubling. + return `\`${col.replace(/`/g, "``")}\``; + } return col; }; const commaSeparatedColumns = indexCols.map(quoteCol).join(", "); diff --git a/core/main_test.ts b/core/main_test.ts index 1e753d11..aff5341a 100644 --- a/core/main_test.ts +++ b/core/main_test.ts @@ -126,7 +126,7 @@ suite("@sqlanvil/core", ({ afterEach }) => { }); suite("warehouse config", () => { - ["bigquery", "postgres", "supabase"].forEach(warehouse => { + ["bigquery", "postgres", "supabase", "mysql"].forEach(warehouse => { test(`accepts warehouse "${warehouse}"`, () => { const projectConfig = workflowSettingsAsProjectConfig( sqlanvil.WorkflowSettings.create({ warehouse, defaultDataset: "d" }) @@ -145,9 +145,9 @@ suite("@sqlanvil/core", ({ afterEach }) => { test("rejects an unknown warehouse instead of silently defaulting", () => { expect(() => workflowSettingsAsProjectConfig( - sqlanvil.WorkflowSettings.create({ warehouse: "mysql", defaultDataset: "d" }) + sqlanvil.WorkflowSettings.create({ warehouse: "snowflake", defaultDataset: "d" }) ) - ).to.throw(/Unsupported warehouse "mysql"/); + ).to.throw(/Unsupported warehouse "snowflake"/); }); }); diff --git a/core/workflow_settings.ts b/core/workflow_settings.ts index 9072931b..64e2071a 100644 --- a/core/workflow_settings.ts +++ b/core/workflow_settings.ts @@ -154,7 +154,7 @@ export function workflowSettingsAsProjectConfig( projectConfig.includeTestsInCompiledGraph = workflowSettings.includeTestsInCompiledGraph; } - const supportedWarehouses = ["bigquery", "postgres", "supabase"]; + const supportedWarehouses = ["bigquery", "postgres", "supabase", "mysql"]; if (workflowSettings.connections) { projectConfig.connections = workflowSettings.connections; From 25c58a400839eb8de35f9496baab96ef65bb09ed Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 09:34:50 -0500 Subject: [PATCH 03/12] build(deps): add mysql2 for MySQL/MariaDB adapter Co-Authored-By: Claude Opus 4.8 --- cli/api/BUILD | 1 + package.json | 3 ++- yarn.lock | 67 ++++++++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 69 insertions(+), 2 deletions(-) diff --git a/cli/api/BUILD b/cli/api/BUILD index e60cc117..2c90d60b 100644 --- a/cli/api/BUILD +++ b/cli/api/BUILD @@ -42,6 +42,7 @@ ts_library( "@npm//google-sql-syntax-ts", "@npm//js-beautify", "@npm//js-yaml", + "@npm//mysql2", "@npm//pg", "@npm//promise-pool-executor", "@npm//protobufjs", diff --git a/package.json b/package.json index e3df4aaf..63d16278 100644 --- a/package.json +++ b/package.json @@ -5,8 +5,8 @@ "@bazel/labs": "^0.42.3", "@bazel/rollup": "^3.0.0", "@bazel/typescript": "^3.0.0", - "@google-cloud/storage": "^7.19.0", "@google-cloud/bigquery": "~8.3.0", + "@google-cloud/storage": "^7.19.0", "@rollup/plugin-node-resolve": "^7.1.3", "@types/chai": "^4.1.7", "@types/diff": "^4.0.2", @@ -46,6 +46,7 @@ "long": "^4.0.0", "minimist": "^1.2.6", "moo": "^0.5.0", + "mysql2": "^3.11.0", "object-sizeof": "^1.6.1", "parse-duration": "^1.0.0", "pg": "^8.11.3", diff --git a/yarn.lock b/yarn.lock index a56c740e..9ff4eb8c 100644 --- a/yarn.lock +++ b/yarn.lock @@ -899,6 +899,11 @@ aws-sign2@~0.7.0: resolved "https://registry.yarnpkg.com/aws-sign2/-/aws-sign2-0.7.0.tgz#b46e890934a9591f2d2f6f86d7e6a9f1b3fe76a8" integrity "sha1-tG6JCTSpWR8tL2+G1+ap8bP+dqg= sha512-08kcGqnYf/YmjoRhfxyu+CLxBjUtHLXLXX/vUfx9l2LYzG3c1m61nrpyFUZI6zeS+Li/wWMMidD9KgrqtGq3mA==" +aws-ssl-profiles@^1.1.2: + version "1.1.2" + resolved "https://registry.yarnpkg.com/aws-ssl-profiles/-/aws-ssl-profiles-1.1.2.tgz#157dd77e9f19b1d123678e93f120e6f193022641" + integrity sha512-NZKeq9AfyQvEeNlN0zSYAaWrmBffJh3IELMZfRpJVWgrpEbtEpnjvzqBPf+mxoI287JohRDoa+/nsfqqiZmF6g== + aws4@^1.8.0: version "1.8.0" resolved "https://registry.yarnpkg.com/aws4/-/aws4-1.8.0.tgz#f0e003d9ca9e7f59c7a508945d7b2ef9a04a542f" @@ -1416,6 +1421,11 @@ denodeify@^1.2.1: resolved "https://registry.yarnpkg.com/denodeify/-/denodeify-1.2.1.tgz#3a36287f5034e699e7577901052c2e6c94251631" integrity "sha1-OjYof1A05pnnV3kBBSwubJQlFjE= sha512-KNTihKNmQENUZeKu5fzfpzRqR5S2VMp4gl9RFHiWzj9DfvYQPMJ6XHKNaQxaGCXwPk6y9yme3aUoaiAe+KX+vg==" +denque@^2.1.0: + version "2.1.0" + resolved "https://registry.yarnpkg.com/denque/-/denque-2.1.0.tgz#e93e1a6569fb5e66f16a3c2a2964617d349d6ab1" + integrity sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw== + diff@^3.2.0: version "3.5.1" resolved "https://registry.yarnpkg.com/diff/-/diff-3.5.1.tgz#e7fae480379d2e944c68ff0f5e1c29b6e28c77ab" @@ -2015,6 +2025,13 @@ gcp-metadata@^8.0.0: google-logging-utils "^1.0.0" json-bigint "^1.0.0" +generate-function@^2.3.1: + version "2.3.1" + resolved "https://registry.yarnpkg.com/generate-function/-/generate-function-2.3.1.tgz#f069617690c10c868e73b8465746764f97c3479f" + integrity sha512-eeB5GfMNeevm/GRYq20ShmsaGcmI81kIX2K9XQx5miC8KdHaC6Jm0qQ8ZNeGOi7wYB8OsdxKs+Y2oVuTFuVwKQ== + dependencies: + is-property "^1.0.2" + get-caller-file@^2.0.5: version "2.0.5" resolved "https://registry.yarnpkg.com/get-caller-file/-/get-caller-file-2.0.5.tgz#4f94412a82db32f36e3b0b9741f8a97feb031f7e" @@ -2356,6 +2373,13 @@ https-proxy-agent@^7.0.1: agent-base "^7.1.2" debug "4" +iconv-lite@^0.7.2: + version "0.7.2" + resolved "https://registry.yarnpkg.com/iconv-lite/-/iconv-lite-0.7.2.tgz#d0bdeac3f12b4835b7359c2ad89c422a4d1cc72e" + integrity sha512-im9DjEDQ55s9fL4EYzOAv0yMqmMBSZp6G0VvFyTMPKWxiSBHUj9NW/qqLmXUwXrrM7AvqSlTCfvqRb0cM8yYqw== + dependencies: + safer-buffer ">= 2.1.2 < 3.0.0" + ieee754@^1.1.4: version "1.1.13" resolved "https://registry.yarnpkg.com/ieee754/-/ieee754-1.1.13.tgz#ec168558e95aa181fd87d37f55c32bbcb6708b84" @@ -2515,6 +2539,11 @@ is-plain-object@^2.0.3, is-plain-object@^2.0.4: dependencies: isobject "^3.0.1" +is-property@^1.0.2: + version "1.0.2" + resolved "https://registry.yarnpkg.com/is-property/-/is-property-1.0.2.tgz#57fe1c4e48474edd65b09911f26b1cd4095dda84" + integrity sha512-Ks/IoX00TtClbGQr4TWXemAnktAQvYB7HzcCxDGqEZU6oCmb2INHuOoKxbtR+HFkmYWBKv/dOZtGRiAjDhj92g== + is-stream@^2.0.0: version "2.0.1" resolved "https://registry.yarnpkg.com/is-stream/-/is-stream-2.0.1.tgz#fac1e3d53b97ad5a9d0ae9cef2389f5810a5c077" @@ -2846,6 +2875,11 @@ long@^5.0.0: resolved "https://registry.yarnpkg.com/long/-/long-5.2.0.tgz#2696dadf4b4da2ce3f6f6b89186085d94d52fd61" integrity sha512-9RTUNjK60eJbx3uz+TEGF7fUr29ZDxR5QzXcyDpeSfeH28S9ycINflOgOlppit5U+4kNTe83KQnMEerw7GmE8w== +long@^5.3.2: + version "5.3.2" + resolved "https://registry.yarnpkg.com/long/-/long-5.3.2.tgz#1d84463095999262d7d7b7f8bfd4a8cc55167f83" + integrity sha512-mNAgZ1GmyNhD7AuqnTG3/VQ26o760+ZYBPKjPvugO8+nLbYfX6TVpJPseBvopbdY+qpZ/lKUnmEc1LeZYS3QAA== + lru-cache@^10.2.0: version "10.4.3" resolved "https://registry.yarnpkg.com/lru-cache/-/lru-cache-10.4.3.tgz#410fc8a17b70e598013df257c2446b7f3383f119" @@ -2871,6 +2905,11 @@ lru-cache@^6.0.0: dependencies: yallist "^4.0.0" +lru.min@^1.1.0, lru.min@^1.1.4: + version "1.1.4" + resolved "https://registry.yarnpkg.com/lru.min/-/lru.min-1.1.4.tgz#6ea1737a8c1ba2300cc87ad46910a4bdffa0117b" + integrity sha512-DqC6n3QQ77zdFpCMASA1a3Jlb64Hv2N2DciFGkO/4L9+q/IpIAuRlKOvCXabtRW6cQf8usbmM6BE/TOPysCdIA== + lunr@^2.3.8: version "2.3.9" resolved "https://registry.npmjs.org/lunr/-/lunr-2.3.9.tgz#18b123142832337dd6e964df1a5a7707b25d35e1" @@ -3068,6 +3107,27 @@ mute-stream@~0.0.4: resolved "https://registry.yarnpkg.com/mute-stream/-/mute-stream-0.0.8.tgz#1630c42b2251ff81e2a283de96a5497ea92e5e0d" integrity "sha1-FjDEKyJR/4HiooPelqVJfqkuXg0= sha512-nnbWWOkoWyUsTjKrhgD0dcz22mdkSnpYqbEjIm2nhwhuxlSkpywJmBo8h0ZqJdkp73mb90SssHkN4rsRaBAfAA==" +mysql2@^3.11.0: + version "3.22.5" + resolved "https://registry.yarnpkg.com/mysql2/-/mysql2-3.22.5.tgz#26c51c035ac577579ad239168015ad1eec321679" + integrity sha512-95uZ2TrPWAZdwpB3vvvDbmEMcNG8yIeNCyu6GUcr/QnWEE/wXm7+mhOCsdQfWQDTV7qYT/PDUZ4U4UPP4AsXqQ== + dependencies: + aws-ssl-profiles "^1.1.2" + denque "^2.1.0" + generate-function "^2.3.1" + iconv-lite "^0.7.2" + long "^5.3.2" + lru.min "^1.1.4" + named-placeholders "^1.1.6" + sql-escaper "^1.3.3" + +named-placeholders@^1.1.6: + version "1.1.6" + resolved "https://registry.yarnpkg.com/named-placeholders/-/named-placeholders-1.1.6.tgz#c50c6920b43f258f59c16add1e56654f5cc02bb5" + integrity sha512-Tz09sEL2EEuv5fFowm419c1+a/jSMiBjI9gHxVLrVdbUkkNUUfjsVYs9pVZu5oCon/kmRh9TfLEObFtkVxmY0w== + dependencies: + lru.min "^1.1.0" + nanomatch@^1.2.9: version "1.2.13" resolved "https://registry.yarnpkg.com/nanomatch/-/nanomatch-1.2.13.tgz#b87a8aa4fc0de8fe6be88895b38983ff265bd119" @@ -3806,7 +3866,7 @@ safe-regex@^1.1.0: dependencies: ret "~0.1.10" -safer-buffer@^2.0.2, safer-buffer@^2.1.0, safer-buffer@~2.1.0: +"safer-buffer@>= 2.1.2 < 3.0.0", safer-buffer@^2.0.2, safer-buffer@^2.1.0, safer-buffer@~2.1.0: version "2.1.2" resolved "https://registry.yarnpkg.com/safer-buffer/-/safer-buffer-2.1.2.tgz#44fa161b0187b9549dd84bb91802f9bd8385cd6a" integrity "sha1-RPoWGwGHuVSd2Eu5GAL5vYOFzWo= sha512-YZo3K82SD7Riyi0E1EQPojLz7kpepnSQI9IyPbHHg1XXXevb5dJI7tpyN2ADxGcQbHG7vcyRHk0cbwqcQriUtg==" @@ -3996,6 +4056,11 @@ sprintf-js@~1.0.2: resolved "https://registry.yarnpkg.com/sprintf-js/-/sprintf-js-1.0.3.tgz#04e6926f662895354f3dd015203633b857297e2c" integrity "sha1-BOaSb2YolTVPPdAVIDYzuFcpfiw= sha512-D9cPgkvLlV3t3IzL0D0YLvGA9Ahk4PcvVwUbN0dSGr1aP0Nrt4AEnTUbuGvquEC0mA64Gqt1fzirlRs5ibXx8g==" +sql-escaper@^1.3.3: + version "1.3.3" + resolved "https://registry.yarnpkg.com/sql-escaper/-/sql-escaper-1.3.3.tgz#65faf89f048d26bb9a75566b82b5990ddf8a5b7f" + integrity sha512-BsTCV265VpTp8tm1wyIm1xqQCS+Q9NHx2Sr+WcnUrgLrQ6yiDIvHYJV5gHxsj1lMBy2zm5twLaZao8Jd+S8JJw== + sshpk@^1.7.0: version "1.16.1" resolved "https://registry.yarnpkg.com/sshpk/-/sshpk-1.16.1.tgz#fb661c0bef29b39db40769ee39fa70093d6f6877" From 5d35cad37ab5d570ab05d648cf647c2444d9f74a Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 09:36:39 -0500 Subject: [PATCH 04/12] feat(mysql): MysqlExecutionSql DDL/DML generator Table CTAS, CREATE OR REPLACE VIEW, incremental (CTAS + unique index then ON DUPLICATE KEY UPDATE upsert), assertion via view + row count. Materialized views error out (deferred). Wired into the ExecutionSql dispatcher. Co-Authored-By: Claude Opus 4.8 --- cli/api/dbadapters/execution_sql.ts | 3 + cli/api/dbadapters/mysql_execution_sql.ts | 128 ++++++++++++++++++++++ cli/api/execution_sql_test.ts | 74 +++++++++++++ 3 files changed, 205 insertions(+) create mode 100644 cli/api/dbadapters/mysql_execution_sql.ts diff --git a/cli/api/dbadapters/execution_sql.ts b/cli/api/dbadapters/execution_sql.ts index 2a063d4e..590ea197 100644 --- a/cli/api/dbadapters/execution_sql.ts +++ b/cli/api/dbadapters/execution_sql.ts @@ -1,4 +1,5 @@ import { BigQueryExecutionSql } from "sa/cli/api/dbadapters/bigquery_execution_sql"; +import { MysqlExecutionSql } from "sa/cli/api/dbadapters/mysql_execution_sql"; import { PostgresExecutionSql } from "sa/cli/api/dbadapters/postgres_execution_sql"; import { concatenateQueries, Tasks } from "sa/cli/api/dbadapters/tasks"; import { ErrorWithCause } from "sa/common/errors/errors"; @@ -36,6 +37,8 @@ export class ExecutionSql implements IExecutionSql { const warehouse = (project.warehouse || "bigquery").toLowerCase(); if (warehouse === "postgres" || warehouse === "supabase") { this.delegate = new PostgresExecutionSql(project, sqlanvilCoreVersion, uniqueIdGenerator); + } else if (warehouse === "mysql") { + this.delegate = new MysqlExecutionSql(project, sqlanvilCoreVersion, uniqueIdGenerator); } else { this.delegate = new BigQueryExecutionSql(project, sqlanvilCoreVersion, uniqueIdGenerator); } diff --git a/cli/api/dbadapters/mysql_execution_sql.ts b/cli/api/dbadapters/mysql_execution_sql.ts new file mode 100644 index 00000000..d84864bb --- /dev/null +++ b/cli/api/dbadapters/mysql_execution_sql.ts @@ -0,0 +1,128 @@ +import { IExecutionSql } from "sa/cli/api/dbadapters/execution_sql"; +import { Task, Tasks } from "sa/cli/api/dbadapters/tasks"; +import { CompilationSql } from "sa/core/compilation_sql"; +import { sqlanvil } from "sa/protos/ts"; + +// MySQL/MariaDB DDL/DML generator. Emits portable MySQL-dialect SQL — the same +// statements run against both engines (engine-specific features ride through +// `operations`). Mirrors PostgresExecutionSql's structure; deliberately omits +// the Postgres-only surface (storage options, partitioning, materialized views, +// COMMENT metadata) — see the adapter design doc for what's deferred. +export class MysqlExecutionSql implements IExecutionSql { + private readonly CompilationSql: CompilationSql; + + constructor( + private readonly project: sqlanvil.IProjectConfig, + private readonly sqlanvilCoreVersion: string, + private readonly uniqueIdGenerator: () => string = () => Math.random().toString(36).substring(2) + ) { + this.CompilationSql = new CompilationSql(project, sqlanvilCoreVersion); + } + + public resolveTarget(target: sqlanvil.ITarget): string { + return this.CompilationSql.resolveTarget(target); + } + + public dropIfExists(target: sqlanvil.ITarget, type: sqlanvil.TableMetadata.Type): string { + if (type === sqlanvil.TableMetadata.Type.VIEW) { + return `drop view if exists ${this.resolveTarget(target)}`; + } + return `drop table if exists ${this.resolveTarget(target)}`; + } + + public publishTasks( + table: sqlanvil.ITable, + runConfig: sqlanvil.IRunConfig, + tableMetadata?: sqlanvil.ITableMetadata + ): Tasks { + const tasks = new Tasks(); + const target = this.resolveTarget(table.target); + + if (table.enumType === sqlanvil.TableType.VIEW) { + if (table.materialized) { + throw new Error( + `Materialized views are not supported on mysql (action ${target}). ` + + `Use a table, or emulate refresh via operations.` + ); + } + // CREATE OR REPLACE VIEW is atomic in MySQL/MariaDB — no drop needed. + tasks.add(Task.statement(`create or replace view ${target} as ${table.query}`)); + return tasks; + } + + if (table.enumType === sqlanvil.TableType.INCREMENTAL) { + const fresh = !this.shouldWriteIncrementally(table, runConfig, tableMetadata); + if (fresh) { + // Full refresh or first build: drop + CTAS, then add the unique index that + // ON DUPLICATE KEY UPDATE relies on for subsequent incremental appends. + tasks.add(Task.statement(this.dropIfExists(table.target, sqlanvil.TableMetadata.Type.TABLE))); + tasks.add(Task.statement(`create table ${target} as ${table.query}`)); + if (table.uniqueKey && table.uniqueKey.length > 0) { + const idx = `uq_${table.target.schema}_${table.target.name}`.slice(0, 63); + const cols = table.uniqueKey.map(k => `\`${k}\``).join(", "); + tasks.add(Task.statement(`alter table ${target} add unique index \`${idx}\` (${cols})`)); + } + } else { + tasks.add(Task.statement(this.upsertInto(table, tableMetadata))); + } + return tasks; + } + + // Plain table: drop + CTAS. + tasks.add(Task.statement(this.dropIfExists(table.target, sqlanvil.TableMetadata.Type.TABLE))); + tasks.add(Task.statement(`create table ${target} as ${table.query}`)); + return tasks; + } + + public assertTasks( + assertion: sqlanvil.IAssertion, + projectConfig: sqlanvil.IProjectConfig + ): Tasks { + // The assertion query is warehouse-agnostic SQL produced by the compiler. + // Mirror the Postgres path: materialize it as a view (catches syntax errors), + // then count rows — any returned row is a failing record. + const tasks = new Tasks(); + const target = this.resolveTarget(assertion.target); + tasks.add(Task.statement(this.dropIfExists(assertion.target, sqlanvil.TableMetadata.Type.VIEW))); + tasks.add(Task.statement(`create or replace view ${target} as ${assertion.query}`)); + tasks.add(Task.assertion(`select sum(1) as row_count from ${target}`)); + return tasks; + } + + private shouldWriteIncrementally( + table: sqlanvil.ITable, + runConfig: sqlanvil.IRunConfig, + tableMetadata?: sqlanvil.ITableMetadata + ): boolean { + return ( + !runConfig.fullRefresh && + !!tableMetadata && + tableMetadata.type === sqlanvil.TableMetadata.Type.TABLE + ); + } + + private getIncrementalQuery(table: sqlanvil.ITable): string { + return table.incrementalQuery || table.query; + } + + private upsertInto(table: sqlanvil.ITable, tableMetadata?: sqlanvil.ITableMetadata): string { + // MySQL dialect: INSERT INTO target (cols) SELECT cols FROM (query) AS insertions + // ON DUPLICATE KEY UPDATE col = values(col), ... — relies on the unique index + // created on first build. + const target = this.resolveTarget(table.target); + const columns = (tableMetadata?.fields || []).map(f => f.name); + const query = this.getIncrementalQuery(table); + if (columns.length === 0) { + return `insert into ${target} select * from (${query}) as insertions`; + } + const backticked = columns.map(c => `\`${c}\``); + const updates = columns + .filter(c => !(table.uniqueKey || []).includes(c)) + .map(c => `\`${c}\` = values(\`${c}\`)`) + .join(", "); + const tail = updates.length > 0 ? ` on duplicate key update ${updates}` : ""; + return `insert into ${target} (${backticked.join(", ")}) select ${backticked.join( + ", " + )} from (${query}) as insertions${tail}`; + } +} diff --git a/cli/api/execution_sql_test.ts b/cli/api/execution_sql_test.ts index 1615c61e..826c7dd7 100644 --- a/cli/api/execution_sql_test.ts +++ b/cli/api/execution_sql_test.ts @@ -456,3 +456,77 @@ suite("ExecutionSql with Postgres/Supabase", () => { }); }); +suite("mysql execution sql", () => { + const project: sqlanvil.IProjectConfig = { warehouse: "mysql" }; + const sql = new ExecutionSql(project, "1.5.0"); + const baseTable = (over: Partial = {}): sqlanvil.ITable => ({ + target: { schema: "db", name: "t" }, + query: "select 1 as id", + enumType: sqlanvil.TableType.TABLE, + ...over + }); + + test("table: drop + CTAS with backticks", () => { + const stmts = sql + .publishTasks(baseTable(), { fullRefresh: false }) + .build() + .map(t => t.statement); + expect(stmts).to.include("drop table if exists `db`.`t`"); + expect(stmts).to.include("create table `db`.`t` as select 1 as id"); + }); + + test("view: CREATE OR REPLACE VIEW", () => { + const stmts = sql + .publishTasks(baseTable({ enumType: sqlanvil.TableType.VIEW }), { fullRefresh: false }) + .build() + .map(t => t.statement); + expect(stmts).to.include("create or replace view `db`.`t` as select 1 as id"); + }); + + test("incremental fresh-create adds a unique index on the uniqueKey", () => { + const stmts = sql + .publishTasks( + baseTable({ enumType: sqlanvil.TableType.INCREMENTAL, uniqueKey: ["id"] }), + { fullRefresh: true } + ) + .build() + .map(t => t.statement); + expect(stmts.some(s => /alter table `db`\.`t` add unique index .* \(`id`\)/.test(s))).to.equal( + true + ); + }); + + test("incremental append upserts via ON DUPLICATE KEY UPDATE", () => { + const stmts = sql + .publishTasks( + baseTable({ + enumType: sqlanvil.TableType.INCREMENTAL, + uniqueKey: ["id"], + incrementalQuery: "select 1 as id" + }), + { fullRefresh: false }, + { + target: { schema: "db", name: "t" }, + type: sqlanvil.TableMetadata.Type.TABLE, + fields: [{ name: "id" }, { name: "v" }] + } + ) + .build() + .map(t => t.statement); + expect( + stmts.some(s => s.includes("on duplicate key update") && s.includes("`v` = values(`v`)")) + ).to.equal(true); + }); + + test("materialized view is rejected on mysql", () => { + expect(() => + sql + .publishTasks( + baseTable({ enumType: sqlanvil.TableType.VIEW, materialized: true }), + { fullRefresh: false } + ) + .build() + ).to.throw(/materialized views are not supported on mysql/i); + }); +}); + From 6a0095059b4829c15e33dba781bfc3454acb554d Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 09:37:47 -0500 Subject: [PATCH 05/12] feat(mysql): MySqlPoolExecutor (release-once, fail-fast verify) mysql2-backed pool with withClientLock + single release path, verifyConnection fail-fast, and a convertFieldType mapping MySQL DATA_TYPEs to field primitives. Co-Authored-By: Claude Opus 4.8 --- cli/api/utils/BUILD | 1 + cli/api/utils/mysql.ts | 120 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+) create mode 100644 cli/api/utils/mysql.ts diff --git a/cli/api/utils/BUILD b/cli/api/utils/BUILD index 5202bc77..98b325bc 100644 --- a/cli/api/utils/BUILD +++ b/cli/api/utils/BUILD @@ -12,6 +12,7 @@ ts_library( "//protos:ts", "@npm//@types/node", "@npm//@types/pg", + "@npm//mysql2", "@npm//object-sizeof", "@npm//pg", "@npm//pg-query-stream", diff --git a/cli/api/utils/mysql.ts b/cli/api/utils/mysql.ts new file mode 100644 index 00000000..d6d6abcb --- /dev/null +++ b/cli/api/utils/mysql.ts @@ -0,0 +1,120 @@ +import * as mysql from "mysql2/promise"; + +import { sqlanvil } from "sa/protos/ts"; + +// Connection-pool lifecycle for MySQL/MariaDB via mysql2. Mirrors PgPoolExecutor: +// fail-fast verifyConnection, a withClientLock that leases a single connection +// for a unit of work, and a single release path (the #32 release-once discipline +// baked in from the start — a double release here would surface a confusing +// pool error ahead of the real query error). +export class MySqlPoolExecutor { + private pool: mysql.Pool; + + constructor(config: mysql.PoolOptions, options?: { concurrencyLimit?: number }) { + this.pool = mysql.createPool({ + ...config, + connectionLimit: options?.concurrencyLimit || 10, + waitForConnections: true, + // Generated SQL runs one statement per task; disabling multi-statement + // execution keeps a single bad statement from chaining unexpected effects. + multipleStatements: false + }); + } + + /** + * Acquire a single connection and run a trivial query to verify the + * credentials/host before any real work fans out. Connecting is where auth + * happens, so a bad password/host fails here with one connection attempt + * rather than N parallel auth failures. + */ + public async verifyConnection(): Promise { + const conn = await this.pool.getConnection(); + try { + await conn.query("select 1"); + } finally { + conn.release(); + } + } + + public async withClientLock( + callback: (client: { + execute(statement: string, options?: { params?: any[]; rowLimit?: number }): Promise; + }) => Promise + ): Promise { + const conn = await this.pool.getConnection(); + // Release exactly once — from the finally below. A second release would trip + // mysql2's pool accounting and mask the real error. + let released = false; + const releaseOnce = () => { + if (released) { + return; + } + released = true; + conn.release(); + }; + try { + return await callback({ + execute: async ( + statement: string, + options: { params?: any[]; rowLimit?: number } = { rowLimit: 1000 } + ): Promise => { + const [rows] = await conn.query(statement, options.params || []); + const arr = Array.isArray(rows) ? (rows as any[]) : []; + return options.rowLimit && arr.length > options.rowLimit + ? arr.slice(0, options.rowLimit) + : arr; + } + }); + } finally { + releaseOnce(); + } + } + + public async close(): Promise { + await this.pool.end(); + } +} + +// Maps MySQL/MariaDB information_schema.columns DATA_TYPE values to sqlanvil +// field primitives. DATA_TYPE excludes the length/precision suffix, so no +// stripping is needed (unlike Postgres's format_type). +export function convertFieldType(type: string) { + switch (String(type).toUpperCase()) { + case "FLOAT": + case "DOUBLE": + case "REAL": + return sqlanvil.Field.Primitive.FLOAT; + case "TINYINT": + case "SMALLINT": + case "MEDIUMINT": + case "INT": + case "INTEGER": + case "BIGINT": + case "YEAR": + return sqlanvil.Field.Primitive.INTEGER; + case "DECIMAL": + case "DEC": + case "NUMERIC": + case "FIXED": + return sqlanvil.Field.Primitive.NUMERIC; + case "BOOL": + case "BOOLEAN": + return sqlanvil.Field.Primitive.BOOLEAN; + case "CHAR": + case "VARCHAR": + case "TINYTEXT": + case "TEXT": + case "MEDIUMTEXT": + case "LONGTEXT": + case "ENUM": + case "SET": + return sqlanvil.Field.Primitive.STRING; + case "DATE": + return sqlanvil.Field.Primitive.DATE; + case "DATETIME": + case "TIMESTAMP": + return sqlanvil.Field.Primitive.TIMESTAMP; + default: + return sqlanvil.Field.Primitive.UNKNOWN; + } +} From d3d9e3ec634d6fb863fe36fa7c20c7a856bf2f80 Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 09:39:01 -0500 Subject: [PATCH 06/12] feat(mysql): MySqlDbAdapter (IDbAdapter via mysql2) create() with fail-fast verify + optional TLS, withClientLock delegating to the pool executor, EXPLAIN-based evaluate, information_schema introspection (tables/table/search/schemas), CREATE DATABASE IF NOT EXISTS, deleteTable. setMetadata is a deferred no-op for the MVP. Co-Authored-By: Claude Opus 4.8 --- cli/api/dbadapters/mysql.ts | 278 ++++++++++++++++++++++++++++++++++++ 1 file changed, 278 insertions(+) create mode 100644 cli/api/dbadapters/mysql.ts diff --git a/cli/api/dbadapters/mysql.ts b/cli/api/dbadapters/mysql.ts new file mode 100644 index 00000000..97592577 --- /dev/null +++ b/cli/api/dbadapters/mysql.ts @@ -0,0 +1,278 @@ +import { collectEvaluationQueries, QueryOrAction } from "sa/cli/api/dbadapters/execution_sql"; +import { + IDbAdapter, + IDbClient, + IExecutionResult, + IExecutionResultRaw, + OnCancel +} from "sa/cli/api/dbadapters/index"; +import { convertFieldType, MySqlPoolExecutor } from "sa/cli/api/utils/mysql"; +import { ErrorWithCause } from "sa/common/errors/errors"; +import { sqlanvil } from "sa/protos/ts"; + +// MySQL/MariaDB has no catalog level above the database, so "schema" and +// "database" are the same thing — these are the engine-managed databases we +// never treat as user schemas. +const INTERNAL_SCHEMAS = new Set([ + "information_schema", + "mysql", + "performance_schema", + "sys" +]); + +export class MySqlDbAdapter implements IDbAdapter { + public static async create( + credentials: sqlanvil.IMysqlConnection, + options?: { concurrencyLimit?: number; disableSslForTestsOnly?: boolean } + ): Promise { + const sslMode = (credentials.sslMode || "").toLowerCase(); + const ssl = + !options?.disableSslForTestsOnly && sslMode && sslMode !== "disable" + ? // Managed MySQL providers serve certs signed by their own CA; skipping + // verification is the documented path for sslmode=require. Stricter + // verification would need a CA bundle we don't ship today. + { rejectUnauthorized: false } + : undefined; + const queryExecutor = new MySqlPoolExecutor( + { + host: credentials.host, + port: credentials.port || 3306, + user: credentials.user, + password: credentials.password, + database: credentials.database || undefined, + ssl + }, + options + ); + // Fail fast on a single connection before any command fans out. + try { + await queryExecutor.verifyConnection(); + } catch (e) { + await queryExecutor.close().catch(() => undefined); + throw new ErrorWithCause( + `Could not connect to MySQL at ${credentials.host}:${credentials.port || 3306} ` + + `as "${credentials.user}": ${e.message}`, + e + ); + } + return new MySqlDbAdapter(queryExecutor); + } + + protected constructor(protected readonly queryExecutor: MySqlPoolExecutor) {} + + public async execute( + statement: string, + options: { + params?: any[]; + onCancel?: OnCancel; + rowLimit?: number; + byteLimit?: number; + includeQueryInError?: boolean; + } = { rowLimit: 1000, byteLimit: 1024 * 1024 } + ): Promise { + return await this.withClientLock(client => client.execute(statement, options)); + } + + public async executeRaw( + statement: string, + options: { + params?: any[]; + rowLimit?: number; + } = { rowLimit: 1000 } + ): Promise { + const result = await this.execute(statement, options); + return { ...result, schema: [] }; + } + + public async withClientLock(callback: (client: IDbClient) => Promise): Promise { + return await this.queryExecutor.withClientLock(client => + callback({ + execute: async ( + stmt: string, + opts: { + params?: any[]; + onCancel?: OnCancel; + rowLimit?: number; + byteLimit?: number; + includeQueryInError?: boolean; + } = { rowLimit: 1000, byteLimit: 1024 * 1024 } + ): Promise => { + try { + const rows = await client.execute(stmt, { params: opts.params, rowLimit: opts.rowLimit }); + return { rows, metadata: {} }; + } catch (e) { + if (opts.includeQueryInError) { + throw new Error(`Error encountered while running "${stmt}": ${e.message}`); + } + throw new ErrorWithCause(`Error executing mysql query: ${e.message}`, e); + } + }, + executeRaw: async ( + stmt: string, + opts: { params?: { [name: string]: any }; rowLimit?: number } = { rowLimit: 1000 } + ): Promise => { + const positional = opts.params ? Object.values(opts.params) : undefined; + const rows = await client.execute(stmt, { params: positional, rowLimit: opts.rowLimit }); + return { rows, schema: [], metadata: {} }; + } + }) + ); + } + + public async evaluate(queryOrAction: QueryOrAction): Promise { + // EXPLAIN parses + plans without executing, catching syntax errors and + // missing tables/columns. + const validationQueries = collectEvaluationQueries(queryOrAction, false, (query: string) => + !!query ? `explain ${query}` : "" + ).map((validationQuery, index) => ({ index, validationQuery })); + const validationQueriesWithoutWrappers = collectEvaluationQueries(queryOrAction, false); + + const queryEvaluations = new Array(); + for (const { index, validationQuery } of validationQueries) { + let evaluationResponse: sqlanvil.IQueryEvaluation = { + status: sqlanvil.QueryEvaluation.QueryEvaluationStatus.SUCCESS + }; + try { + await this.execute(validationQuery.query); + } catch (e) { + evaluationResponse = { + status: sqlanvil.QueryEvaluation.QueryEvaluationStatus.FAILURE, + error: sqlanvil.QueryEvaluationError.create({ + message: e?.message ? String(e.message) : String(e) + }) + }; + } + queryEvaluations.push( + sqlanvil.QueryEvaluation.create({ + ...evaluationResponse, + incremental: validationQuery.incremental, + query: validationQueriesWithoutWrappers[index].query + }) + ); + } + return queryEvaluations; + } + + public async tables(_database: string, schema?: string): Promise { + const params: any[] = []; + let schemaClause = ""; + if (schema) { + schemaClause = "and table_schema = ?"; + params.push(schema); + } + const queryResult = await this.execute( + `select table_name, table_schema + from information_schema.tables + where table_schema not in ('information_schema', 'mysql', 'performance_schema', 'sys') + ${schemaClause}`, + { params, rowLimit: 10000, includeQueryInError: true } + ); + const targets = queryResult.rows.map(row => ({ + schema: row.table_schema as string, + name: row.table_name as string + })); + return await Promise.all(targets.map(target => this.table(target))); + } + + public async search( + searchText: string, + options: { limit: number } = { limit: 1000 } + ): Promise { + const results = await this.execute( + `select tables.table_schema as table_schema, tables.table_name as table_name + from information_schema.tables as tables + left join information_schema.columns columns + on tables.table_schema = columns.table_schema + and tables.table_name = columns.table_name + where tables.table_schema like ? + or tables.table_name like ? + or columns.column_name like ? + group by 1, 2`, + { + params: [`%${searchText}%`, `%${searchText}%`, `%${searchText}%`], + rowLimit: options.limit + } + ); + return await Promise.all( + results.rows.map(row => + this.table({ + schema: row.table_schema, + name: row.table_name + }) + ) + ); + } + + public async table(target: sqlanvil.ITarget): Promise { + const params = [target.schema, target.name]; + const [tableResults, columnResults] = await Promise.all([ + this.execute( + `select table_type from information_schema.tables + where table_schema = ? and table_name = ?`, + { params, includeQueryInError: true } + ), + this.execute( + `select column_name, data_type, ordinal_position + from information_schema.columns + where table_schema = ? and table_name = ? + order by ordinal_position`, + { params, includeQueryInError: true } + ) + ]); + + if (tableResults.rows.length === 0) { + return null; + } + + // mysql2 returns information_schema column names in their canonical + // upper/lower case depending on server config; normalise via lower-cased keys. + const tableType = String( + tableResults.rows[0].table_type ?? tableResults.rows[0].TABLE_TYPE + ).toUpperCase(); + return sqlanvil.TableMetadata.create({ + target, + type: tableType === "VIEW" ? sqlanvil.TableMetadata.Type.VIEW : sqlanvil.TableMetadata.Type.TABLE, + fields: columnResults.rows.map(row => + sqlanvil.Field.create({ + name: (row.column_name ?? row.COLUMN_NAME) as string, + primitive: convertFieldType((row.data_type ?? row.DATA_TYPE) as string) + }) + ) + }); + } + + public async deleteTable(target: sqlanvil.ITarget): Promise { + const metadata = await this.table(target); + if (!metadata) { + return; + } + const kind = metadata.type === sqlanvil.TableMetadata.Type.VIEW ? "view" : "table"; + await this.execute(`drop ${kind} if exists \`${target.schema}\`.\`${target.name}\``, { + includeQueryInError: true + }); + } + + public async schemas(_database: string): Promise { + const result = await this.execute(`select schema_name from information_schema.schemata`, { + includeQueryInError: true + }); + return result.rows + .map(row => (row.schema_name ?? row.SCHEMA_NAME) as string) + .filter(name => !INTERNAL_SCHEMAS.has(name)); + } + + public async createSchema(_database: string, schema: string): Promise { + await this.execute(`create database if not exists \`${schema}\``, { + includeQueryInError: true + }); + } + + public async setMetadata(_action: sqlanvil.IExecutionAction): Promise { + // Deferred for the MVP — table/column COMMENT metadata is a follow-up PR. + return; + } + + public async close(): Promise { + await this.queryExecutor.close(); + } +} From b46c146effe60886eeb52348ffccc406c9c0bc2c Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 09:40:18 -0500 Subject: [PATCH 07/12] feat(mysql): credentials validation + CLI wiring - credentials.read() validates MysqlConnection (host required) - init scaffolds a mysql .df-credentials.json template - warehouseOption accepts mysql; run/test commands create MySqlDbAdapter Co-Authored-By: Claude Opus 4.8 --- cli/api/commands/credentials.ts | 7 +++++++ cli/api/commands/init.ts | 18 ++++++++++++++++-- cli/index.ts | 7 ++++++- 3 files changed, 29 insertions(+), 3 deletions(-) diff --git a/cli/api/commands/credentials.ts b/cli/api/commands/credentials.ts index 5509dd80..b870edd5 100644 --- a/cli/api/commands/credentials.ts +++ b/cli/api/commands/credentials.ts @@ -21,6 +21,13 @@ export function read(credentialsPath: string, warehouse: string = "bigquery"): a // map in workflow_settings.yaml). It is not part of the write-warehouse // connection, so exclude it from the strict warehouse-credentials validation. const { connections, ...warehouseCredentials } = credentialsAsJson; + if (warehouse.toLowerCase() === "mysql") { + const credentials = verifyObjectMatchesProto(sqlanvil.MysqlConnection, warehouseCredentials); + if (!credentials.host) { + throw new Error(`Error reading credentials file: the host field is required`); + } + return credentials; + } const isPostgres = warehouse.toLowerCase() === "postgres" || warehouse.toLowerCase() === "supabase"; if (isPostgres) { const credentials = verifyObjectMatchesProto(sqlanvil.PostgresConnection, warehouseCredentials); diff --git a/cli/api/commands/init.ts b/cli/api/commands/init.ts index 8b5fd423..703787c8 100644 --- a/cli/api/commands/init.ts +++ b/cli/api/commands/init.ts @@ -32,6 +32,20 @@ function postgresCredentialsTemplate(warehouse: string): string { return `${JSON.stringify(template, null, 2)}\n`; } +// A starter MysqlConnection (strict JSON — no comment keys). Points at a local +// MySQL/MariaDB instance with SSL disabled by default. +function mysqlCredentialsTemplate(): string { + const template = { + host: "localhost", + port: 3306, + database: "sqlanvil", + user: "root", + password: "", + sslMode: "disable" + }; + return `${JSON.stringify(template, null, 2)}\n`; +} + export async function init( projectDir: string, projectConfig: sqlanvil.IProjectConfig @@ -100,13 +114,13 @@ export async function init( fs.writeFileSync(gitignorePath, gitIgnoreContents); filesWritten.push(gitignorePath); - // Postgres/Supabase: scaffold a credentials template (the connection lives in a separate, + // Postgres/Supabase/MySQL: scaffold a credentials template (the connection lives in a separate, // gitignored file — not in workflow_settings.yaml). BigQuery credentials come from gcloud / a // BigQuery key, so no template is written for it. if (!isBigQuery) { fs.writeFileSync( path.join(projectDir, CREDENTIALS_FILENAME), - postgresCredentialsTemplate(warehouse) + warehouse === "mysql" ? mysqlCredentialsTemplate() : postgresCredentialsTemplate(warehouse) ); filesWritten.push(path.join(projectDir, CREDENTIALS_FILENAME)); } diff --git a/cli/index.ts b/cli/index.ts index 73289907..a60ecea1 100644 --- a/cli/index.ts +++ b/cli/index.ts @@ -10,6 +10,7 @@ import { CREDENTIALS_FILENAME } from "sa/cli/api/commands/credentials"; import { assertConnectionCredentialsAvailable } from "sa/cli/api/commands/connection_credentials"; import { IDbAdapter } from "sa/cli/api/dbadapters"; import { BigQueryDbAdapter } from "sa/cli/api/dbadapters/bigquery"; +import { MySqlDbAdapter } from "sa/cli/api/dbadapters/mysql"; import { PostgresDbAdapter } from "sa/cli/api/dbadapters/postgres"; import { SupabaseDbAdapter } from "sa/cli/api/dbadapters/supabase"; import { prettyJsonStringify } from "sa/cli/api/utils"; @@ -326,7 +327,7 @@ const icebergOption = option("iceberg", { const warehouseOption = option("warehouse", { describe: "Target warehouse for the new project.", type: "string", - choices: ["bigquery", "postgres", "supabase"], + choices: ["bigquery", "postgres", "supabase", "mysql"], default: "supabase" }); @@ -664,6 +665,8 @@ export function runCli() { dbadapter = await SupabaseDbAdapter.create(readCredentials); } else if (warehouse.toLowerCase() === "postgres") { dbadapter = await PostgresDbAdapter.create(readCredentials); + } else if (warehouse.toLowerCase() === "mysql") { + dbadapter = await MySqlDbAdapter.create(readCredentials); } else { dbadapter = new BigQueryDbAdapter(readCredentials); } @@ -743,6 +746,8 @@ export function runCli() { dbadapter = await SupabaseDbAdapter.create(readCredentials); } else if (warehouse.toLowerCase() === "postgres") { dbadapter = await PostgresDbAdapter.create(readCredentials); + } else if (warehouse.toLowerCase() === "mysql") { + dbadapter = await MySqlDbAdapter.create(readCredentials); } else { dbadapter = new BigQueryDbAdapter(readCredentials); } From 15eebf9ceddc47a07ce78d70ff96fef836bb54eb Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 10:20:08 -0500 Subject: [PATCH 08/12] test(mysql): docker fixtures for mysql:8 + mariadb:11 MysqlFixture (2-arg ctor, env-bypass for the docker-bazel path; same class serves MySQL 3306 and MariaDB 3307) + run-mysql-db.sh launcher + BUILD. Co-Authored-By: Claude Opus 4.8 --- tools/mysql/BUILD | 14 ++++++ tools/mysql/mysql_fixture.ts | 98 ++++++++++++++++++++++++++++++++++++ tools/mysql/run-mysql-db.sh | 21 ++++++++ 3 files changed, 133 insertions(+) create mode 100644 tools/mysql/BUILD create mode 100644 tools/mysql/mysql_fixture.ts create mode 100755 tools/mysql/run-mysql-db.sh diff --git a/tools/mysql/BUILD b/tools/mysql/BUILD new file mode 100644 index 00000000..49760124 --- /dev/null +++ b/tools/mysql/BUILD @@ -0,0 +1,14 @@ +package(default_visibility = ["//visibility:public"]) + +load("//tools:ts_library.bzl", "ts_library") + +ts_library( + name = "mysql", + srcs = glob(["*.ts"]), + deps = [ + "//common/promises", + "//testing", + "@npm//@types/node", + "@npm//mysql2", + ], +) diff --git a/tools/mysql/mysql_fixture.ts b/tools/mysql/mysql_fixture.ts new file mode 100644 index 00000000..147a88c5 --- /dev/null +++ b/tools/mysql/mysql_fixture.ts @@ -0,0 +1,98 @@ +import { execSync } from "child_process"; +import * as mysql from "mysql2/promise"; + +import { sleepUntil } from "sa/common/promises"; +import { IHookHandler } from "sa/testing"; + +const USE_CLOUD_BUILD_NETWORK = !!process.env.USE_CLOUD_BUILD_NETWORK; +const DOCKER_CONTAINER_NAME = "mysql-sa-integration-testing"; +const MYSQL_IMAGE = "mysql:8"; +const MYSQL_SERVE_PORT = 3306; + +// When MYSQL_HOST is set (the docker-bazel path), the fixture connects to a +// host-provided endpoint instead of booting its own container — exactly like +// PostgresFixture's bypass. This is also how the same class serves both MySQL +// (port 3306) and MariaDB (port 3307): engine/port come from env. +function isDockerBypassed() { + return !!process.env.MYSQL_HOST; +} + +export class MysqlFixture { + public static get host() { + return isDockerBypassed() + ? process.env.MYSQL_HOST || "localhost" + : USE_CLOUD_BUILD_NETWORK + ? DOCKER_CONTAINER_NAME + : "localhost"; + } + + public static get port() { + return isDockerBypassed() + ? process.env.MYSQL_PORT + ? parseInt(process.env.MYSQL_PORT, 10) + : 3306 + : 3306; + } + + public static get user() { + return isDockerBypassed() ? process.env.MYSQL_USER || "root" : "root"; + } + + public static get password() { + return isDockerBypassed() ? process.env.MYSQL_PASSWORD || "password" : "password"; + } + + public static get database() { + return isDockerBypassed() ? process.env.MYSQL_DATABASE || "sqlanvil" : "sqlanvil"; + } + + constructor(setUp: IHookHandler, tearDown: IHookHandler) { + setUp("starting mysql", async () => { + const bypass = isDockerBypassed(); + + if (!bypass) { + execSync( + [ + "docker run", + "--rm", + `--name ${DOCKER_CONTAINER_NAME}`, + "-e MYSQL_ROOT_PASSWORD=password", + "-e MYSQL_DATABASE=sqlanvil", + "-d", + `-p ${MysqlFixture.port}:${MYSQL_SERVE_PORT}`, + USE_CLOUD_BUILD_NETWORK ? "--network cloudbuild" : "", + MYSQL_IMAGE + ].join(" ") + ); + } + + // Block until mysql is ready to accept connections (the server takes a few + // seconds to initialize even after the container is up). + await sleepUntil(async () => { + let conn: mysql.Connection | undefined; + try { + conn = await mysql.createConnection({ + host: MysqlFixture.host, + port: MysqlFixture.port, + user: MysqlFixture.user, + password: MysqlFixture.password + }); + await conn.query("select 1"); + return true; + } catch (e) { + return false; + } finally { + if (conn) { + await conn.end().catch(() => undefined); + } + } + }, 500); + }); + + tearDown("stopping mysql", () => { + if (!isDockerBypassed()) { + execSync(`docker stop ${DOCKER_CONTAINER_NAME}`); + } + }); + } +} diff --git a/tools/mysql/run-mysql-db.sh b/tools/mysql/run-mysql-db.sh new file mode 100755 index 00000000..654731b1 --- /dev/null +++ b/tools/mysql/run-mysql-db.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +set -euo pipefail +docker rm -f mysql-sa-itest mariadb-sa-itest 2>/dev/null || true +echo "Launching mysql:8 on 3306 and mariadb:11 on 3307..." +docker run --rm --name mysql-sa-itest -e MYSQL_ROOT_PASSWORD=password -e MYSQL_DATABASE=sqlanvil -p 3306:3306 -d mysql:8 +docker run --rm --name mariadb-sa-itest -e MARIADB_ROOT_PASSWORD=password -e MARIADB_DATABASE=sqlanvil -p 3307:3306 -d mariadb:11 +cat <<'NOTE' +Run the integration spec against each engine (reach containers as host.docker.internal): + + # MySQL + MYSQL_HOST=host.docker.internal MYSQL_PORT=3306 MYSQL_USER=root MYSQL_PASSWORD=password MYSQL_DATABASE=sqlanvil \ + ./scripts/docker-bazel test //tests/integration:mysql.spec \ + --test_env=MYSQL_HOST --test_env=MYSQL_PORT --test_env=MYSQL_USER --test_env=MYSQL_PASSWORD --test_env=MYSQL_DATABASE \ + --jobs=2 --local_ram_resources=2048 + + # MariaDB (point the same env at port 3307) + MYSQL_HOST=host.docker.internal MYSQL_PORT=3307 MYSQL_USER=root MYSQL_PASSWORD=password MYSQL_DATABASE=sqlanvil \ + ./scripts/docker-bazel test //tests/integration:mysql.spec \ + --test_env=MYSQL_HOST --test_env=MYSQL_PORT --test_env=MYSQL_USER --test_env=MYSQL_PASSWORD --test_env=MYSQL_DATABASE \ + --jobs=2 --local_ram_resources=2048 +NOTE From 864568f65786fd11a9187f84d9d82b0421ed8b61 Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 10:49:40 -0500 Subject: [PATCH 09/12] test(mysql): integration spec against mysql:8 and mariadb:11 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Full-project run (table/view/incremental/merge/assertions), incremental append + upsert idempotency, EXPLAIN-based evaluate, fail-fast bad creds — verified PASSED on both mysql:8 (3306) and mariadb:11 (3307). Adds MYSQL_* env forwarding to scripts/docker-bazel; upsertInto only emits ON DUPLICATE KEY UPDATE when a uniqueKey exists (plain append otherwise). Co-Authored-By: Claude Opus 4.8 --- cli/api/dbadapters/mysql_execution_sql.ts | 15 +- scripts/docker-bazel | 5 + tests/integration/BUILD | 26 +++ tests/integration/mysql.spec.ts | 221 ++++++++++++++++++ tests/integration/mysql_project/BUILD | 17 ++ .../definitions/example_assertion_fail.sqlx | 8 + .../definitions/example_assertion_pass.sqlx | 8 + .../definitions/example_incremental.sqlx | 14 ++ .../example_incremental_merge.sqlx | 16 ++ .../definitions/example_table.sqlx | 5 + .../definitions/example_view.sqlx | 5 + .../definitions/sample_data.sqlx | 7 + .../definitions/sample_data_2.sqlx | 7 + tests/integration/mysql_project/package.json | 1 + .../mysql_project/workflow_settings.yaml | 5 + 15 files changed, 356 insertions(+), 4 deletions(-) create mode 100644 tests/integration/mysql.spec.ts create mode 100644 tests/integration/mysql_project/BUILD create mode 100644 tests/integration/mysql_project/definitions/example_assertion_fail.sqlx create mode 100644 tests/integration/mysql_project/definitions/example_assertion_pass.sqlx create mode 100644 tests/integration/mysql_project/definitions/example_incremental.sqlx create mode 100644 tests/integration/mysql_project/definitions/example_incremental_merge.sqlx create mode 100644 tests/integration/mysql_project/definitions/example_table.sqlx create mode 100644 tests/integration/mysql_project/definitions/example_view.sqlx create mode 100644 tests/integration/mysql_project/definitions/sample_data.sqlx create mode 100644 tests/integration/mysql_project/definitions/sample_data_2.sqlx create mode 100644 tests/integration/mysql_project/package.json create mode 100644 tests/integration/mysql_project/workflow_settings.yaml diff --git a/cli/api/dbadapters/mysql_execution_sql.ts b/cli/api/dbadapters/mysql_execution_sql.ts index d84864bb..1f645196 100644 --- a/cli/api/dbadapters/mysql_execution_sql.ts +++ b/cli/api/dbadapters/mysql_execution_sql.ts @@ -116,10 +116,17 @@ export class MysqlExecutionSql implements IExecutionSql { return `insert into ${target} select * from (${query}) as insertions`; } const backticked = columns.map(c => `\`${c}\``); - const updates = columns - .filter(c => !(table.uniqueKey || []).includes(c)) - .map(c => `\`${c}\` = values(\`${c}\`)`) - .join(", "); + // Only an upsert when a uniqueKey exists — the unique index created on first + // build is what ON DUPLICATE KEY UPDATE matches against. Without it this is a + // plain append (the inert clause would never fire anyway). + const uniqueKey = table.uniqueKey || []; + const updates = + uniqueKey.length > 0 + ? columns + .filter(c => !uniqueKey.includes(c)) + .map(c => `\`${c}\` = values(\`${c}\`)`) + .join(", ") + : ""; const tail = updates.length > 0 ? ` on duplicate key update ${updates}` : ""; return `insert into ${target} (${backticked.join(", ")}) select ${backticked.join( ", " diff --git a/scripts/docker-bazel b/scripts/docker-bazel index 4e0f829c..dcc0f45c 100755 --- a/scripts/docker-bazel +++ b/scripts/docker-bazel @@ -48,6 +48,11 @@ docker run --rm "${TTY_FLAGS[@]}" \ -e SUPABASE_PASSWORD \ -e SUPABASE_DATABASE \ -e SUPABASE_CONNECTION_STRING \ + -e MYSQL_HOST \ + -e MYSQL_PORT \ + -e MYSQL_USER \ + -e MYSQL_PASSWORD \ + -e MYSQL_DATABASE \ -v "$REPO_ROOT:/workspace" \ -v sqlanvil-bazel-cache:/root/.cache/bazel \ -v sqlanvil-bazel-disk:/root/.cache/bazel-disk \ diff --git a/tests/integration/BUILD b/tests/integration/BUILD index 7c09b5e3..5197a48b 100644 --- a/tests/integration/BUILD +++ b/tests/integration/BUILD @@ -71,6 +71,32 @@ ts_test_suite( ], ) +ts_test_suite( + name = "mysql_tests", + srcs = [ + "mysql.spec.ts", + ], + data = [ + "//tests/integration/mysql_project:files", + "//tests/integration/mysql_project:node_modules", + ], + tags = ["integration"], + deps = [ + ":utils", + "//cli/api", + "//cli/api/utils", + "//common/promises", + "//core", + "//protos:ts", + "//testing", + "//tools/mysql", + "@npm//@types/chai", + "@npm//@types/long", + "@npm//@types/node", + "@npm//chai", + ], +) + ts_test_suite( name = "supabase_tests", srcs = [ diff --git a/tests/integration/mysql.spec.ts b/tests/integration/mysql.spec.ts new file mode 100644 index 00000000..58b1da25 --- /dev/null +++ b/tests/integration/mysql.spec.ts @@ -0,0 +1,221 @@ +import { expect } from "chai"; + +import * as dfapi from "sa/cli/api"; +import * as dbadapters from "sa/cli/api/dbadapters"; +import { ExecutionSql } from "sa/cli/api/dbadapters/execution_sql"; +import { MySqlDbAdapter } from "sa/cli/api/dbadapters/mysql"; +import { targetAsReadableString } from "sa/core/targets"; +import { sqlanvil } from "sa/protos/ts"; +import { suite, test } from "sa/testing"; +import { compile, getTableRows, keyBy } from "sa/tests/integration/utils"; +import { MysqlFixture } from "sa/tools/mysql/mysql_fixture"; + +// Runs against whichever engine the MYSQL_* env points at — mysql:8 on 3306 or +// mariadb:11 on 3307 (see tools/mysql/run-mysql-db.sh). The generated SQL is the +// same for both, which is the point. +suite("@sqlanvil/integration/mysql", { parallel: false }, ({ before, after }) => { + let dbadapter: dbadapters.IDbAdapter; + + const mysql = new MysqlFixture(before, after); + + // MySQL has no catalog level, so each test "schema" is its own database. + const TEST_DATABASES = [ + "sa_integration_test_project_e2e", + "sa_integration_test_assertions_project_e2e", + "sa_integration_test_direct" + ]; + + before("create adapter", async () => { + dbadapter = await MySqlDbAdapter.create( + { + host: MysqlFixture.host, + port: MysqlFixture.port, + database: MysqlFixture.database, + user: MysqlFixture.user, + password: MysqlFixture.password + }, + { disableSslForTestsOnly: true } + ); + // Clear any stale test databases from previous runs. + for (const database of TEST_DATABASES) { + try { + await dbadapter.execute(`drop database if exists \`${database}\``); + } catch (e) { + // ignore + } + } + }); + + after("cleanup", async () => { + for (const database of TEST_DATABASES) { + try { + await dbadapter.execute(`drop database if exists \`${database}\``); + } catch (e) { + // ignore + } + } + await (dbadapter as MySqlDbAdapter).close(); + }); + + test("create() fails fast with a clear error on bad credentials", { timeout: 30000 }, async () => { + let err: Error | undefined; + try { + await MySqlDbAdapter.create( + { + host: MysqlFixture.host, + port: MysqlFixture.port, + database: MysqlFixture.database, + user: MysqlFixture.user, + password: "definitely-the-wrong-password" + }, + { disableSslForTestsOnly: true } + ); + } catch (e) { + err = e; + } + expect(err, "create() should reject when credentials are bad").to.be.an("error"); + expect(err.message.toLowerCase()).to.match(/could not connect|access denied|authentication/); + }); + + test("a failing statement rejects with the real error", async () => { + let err: Error | undefined; + try { + await dbadapter.execute("selct 1"); + } catch (e) { + err = e; + } + expect(err, "a bad statement should reject").to.be.an("error"); + expect(err.message.toLowerCase()).to.match(/sql syntax|you have an error/); + }); + + test("table and view generate two-part backticked DDL and are queryable", { timeout: 30000 }, async () => { + const database = "sa_integration_test_direct"; + await dbadapter.execute(`create database if not exists \`${database}\``); + const adapter = new ExecutionSql({ warehouse: "mysql" }, "2.0.0"); + + const table: sqlanvil.ITable = { + enumType: sqlanvil.TableType.TABLE, + target: { schema: database, name: "t" }, + query: "select 1 as id union all select 2 as id" + }; + for (const task of adapter.publishTasks(table, { fullRefresh: true }).build()) { + await dbadapter.execute(task.statement); + } + const tableRows = await getTableRows(table.target, adapter, dbadapter); + expect(tableRows.length).to.equal(2); + + const view: sqlanvil.ITable = { + enumType: sqlanvil.TableType.VIEW, + target: { schema: database, name: "v" }, + query: `select id from \`${database}\`.\`t\`` + }; + for (const task of adapter.publishTasks(view, { fullRefresh: false }).build()) { + await dbadapter.execute(task.statement); + } + const viewRows = await getTableRows(view.target, adapter, dbadapter); + expect(viewRows.length).to.equal(2); + + // The adapter introspects the database via information_schema. + const meta = await dbadapter.table({ schema: database, name: "t" }); + expect(meta).to.not.equal(null); + expect(meta.type).to.equal(sqlanvil.TableMetadata.Type.TABLE); + expect(meta.fields.map(f => f.name)).to.include("id"); + + const viewMeta = await dbadapter.table({ schema: database, name: "v" }); + expect(viewMeta.type).to.equal(sqlanvil.TableMetadata.Type.VIEW); + + await dbadapter.execute(`drop database if exists \`${database}\``); + }); + + test("run: full project build, incremental append, assertion pass/fail", { timeout: 120000 }, async () => { + const compiledGraph = await compile("tests/integration/mysql_project", "project_e2e"); + + // Run the whole project. + let executionGraph = await dfapi.build(compiledGraph, {}, dbadapter); + let executedGraph = await dfapi.run(dbadapter, executionGraph).result(); + + const actionMap = keyBy(executedGraph.actions, v => targetAsReadableString(v.target)); + expect(Object.keys(actionMap).length).to.equal(8); + + const expectedFailedActions = [ + "sa_integration_test_assertions_project_e2e.example_assertion_fail" + ]; + for (const actionName of Object.keys(actionMap)) { + const expectedResult = expectedFailedActions.includes(actionName) + ? sqlanvil.ActionResult.ExecutionStatus.FAILED + : sqlanvil.ActionResult.ExecutionStatus.SUCCESSFUL; + expect(actionMap[actionName].status).to.equal( + expectedResult, + `${actionName}: ${actionMap[actionName].tasks.map(t => t.errorMessage).join("\n")}` + ); + } + + // The failing assertion reports through the mysql error prefix. + expect( + actionMap["sa_integration_test_assertions_project_e2e.example_assertion_fail"].tasks.slice(-1)[0] + .errorMessage + ).to.equal("mysql error: Assertion failed: query returned 1 row(s)."); + + const adapter = new ExecutionSql(compiledGraph.projectConfig, compiledGraph.sqlanvilCoreVersion); + + // Incremental (no uniqueKey): 3 rows on first build. + let incremental = keyBy(compiledGraph.tables, t => targetAsReadableString(t.target))[ + "sa_integration_test_project_e2e.example_incremental" + ]; + expect((await getTableRows(incremental.target, adapter, dbadapter)).length).to.equal(3); + + // Incremental merge (uniqueKey): 2 rows on first build. + let merge = keyBy(compiledGraph.tables, t => targetAsReadableString(t.target))[ + "sa_integration_test_project_e2e.example_incremental_merge" + ]; + expect((await getTableRows(merge.target, adapter, dbadapter)).length).to.equal(2); + + // Re-run the tables: incremental appends, merge upserts (no new rows). + executionGraph = await dfapi.build( + compiledGraph, + { actions: ["example_incremental", "example_incremental_merge", "example_table", "example_view"] }, + dbadapter + ); + executedGraph = await dfapi.run(dbadapter, executionGraph).result(); + expect(executedGraph.status).to.equal( + sqlanvil.RunResult.ExecutionStatus.SUCCESSFUL, + executedGraph.actions + .map(action => action.tasks.map(task => task.errorMessage).join("\n")) + .join("\n") + ); + + // Append added 2 rows (user_timestamp > MIN) -> 5 total. + incremental = keyBy(compiledGraph.tables, t => targetAsReadableString(t.target))[ + "sa_integration_test_project_e2e.example_incremental" + ]; + expect((await getTableRows(incremental.target, adapter, dbadapter)).length).to.equal(5); + + // Merge upserted the same two keys -> still 2 rows, values updated to 'new'. + merge = keyBy(compiledGraph.tables, t => targetAsReadableString(t.target))[ + "sa_integration_test_project_e2e.example_incremental_merge" + ]; + const mergeRows = await getTableRows(merge.target, adapter, dbadapter); + expect(mergeRows.length).to.equal(2); + expect(mergeRows.every((r: any) => r.val === "new")).to.equal(true); + }); + + test("evaluate validates good and bad queries via EXPLAIN", { timeout: 30000 }, async () => { + const good = await dbadapter.evaluate( + sqlanvil.Table.create({ + enumType: sqlanvil.TableType.TABLE, + query: "select 1 as id", + target: { schema: "sa_integration_test_direct", name: "ev_ok" } + }) + ); + expect(good[0].status).to.equal(sqlanvil.QueryEvaluation.QueryEvaluationStatus.SUCCESS); + + const bad = await dbadapter.evaluate( + sqlanvil.Table.create({ + enumType: sqlanvil.TableType.TABLE, + query: "thisisillegal", + target: { schema: "sa_integration_test_direct", name: "ev_bad" } + }) + ); + expect(bad[0].status).to.equal(sqlanvil.QueryEvaluation.QueryEvaluationStatus.FAILURE); + }); +}); diff --git a/tests/integration/mysql_project/BUILD b/tests/integration/mysql_project/BUILD new file mode 100644 index 00000000..e83056bb --- /dev/null +++ b/tests/integration/mysql_project/BUILD @@ -0,0 +1,17 @@ +package(default_visibility = ["//tests:__subpackages__"]) + +load("//tools:node_modules.bzl", "node_modules") + +filegroup( + name = "files", + srcs = glob([ + "**/*.*", + ]), +) + +node_modules( + name = "node_modules", + deps = [ + "//packages/@sqlanvil/core:package_tar", + ], +) diff --git a/tests/integration/mysql_project/definitions/example_assertion_fail.sqlx b/tests/integration/mysql_project/definitions/example_assertion_fail.sqlx new file mode 100644 index 00000000..b8af03dd --- /dev/null +++ b/tests/integration/mysql_project/definitions/example_assertion_fail.sqlx @@ -0,0 +1,8 @@ +config { + type: "assertion" +} + +WITH base AS ( + SELECT val1, SUM(1) as row_count FROM ${ref("sample_data_2")} GROUP BY val1 +) +SELECT * FROM base WHERE row_count > 1 diff --git a/tests/integration/mysql_project/definitions/example_assertion_pass.sqlx b/tests/integration/mysql_project/definitions/example_assertion_pass.sqlx new file mode 100644 index 00000000..2b70a4ad --- /dev/null +++ b/tests/integration/mysql_project/definitions/example_assertion_pass.sqlx @@ -0,0 +1,8 @@ +config { + type: "assertion" +} + +WITH base AS ( + SELECT val1, val2, SUM(1) as row_count FROM ${ref("sample_data_2")} GROUP BY val1, val2 +) +SELECT * FROM base WHERE row_count > 1 diff --git a/tests/integration/mysql_project/definitions/example_incremental.sqlx b/tests/integration/mysql_project/definitions/example_incremental.sqlx new file mode 100644 index 00000000..e640337d --- /dev/null +++ b/tests/integration/mysql_project/definitions/example_incremental.sqlx @@ -0,0 +1,14 @@ +config { + type: "incremental" +} + +WITH example_data AS ( + SELECT 1502920304 AS user_timestamp, 3940 AS user_id UNION ALL + SELECT 1502930293 AS user_timestamp, 20492 AS user_id UNION ALL + SELECT 1502940292 AS user_timestamp, 30920 AS user_id +) + +SELECT user_timestamp, user_id +FROM example_data + +${ when(incremental(), `WHERE user_timestamp > (SELECT MIN(user_timestamp) FROM example_data)`) } diff --git a/tests/integration/mysql_project/definitions/example_incremental_merge.sqlx b/tests/integration/mysql_project/definitions/example_incremental_merge.sqlx new file mode 100644 index 00000000..204c360a --- /dev/null +++ b/tests/integration/mysql_project/definitions/example_incremental_merge.sqlx @@ -0,0 +1,16 @@ +config { + type: "incremental", + uniqueKey: ["id_1", "id_2"] +} + +WITH example_data AS ( + SELECT 1 AS ts, 21 AS id_1, 31 AS id_2, 'original' AS val UNION ALL + SELECT 2 AS ts, 22 AS id_1, 32 AS id_2, 'original' AS val UNION ALL + SELECT 3 AS ts, 21 AS id_1, 31 AS id_2, 'new' AS val UNION ALL + SELECT 4 AS ts, 22 AS id_1, 32 AS id_2, 'new' AS val +) + +SELECT ts, id_1, id_2, val +FROM example_data + +${ when(incremental(), `WHERE ts > 2`, `WHERE ts <= 2`) } diff --git a/tests/integration/mysql_project/definitions/example_table.sqlx b/tests/integration/mysql_project/definitions/example_table.sqlx new file mode 100644 index 00000000..6b37db73 --- /dev/null +++ b/tests/integration/mysql_project/definitions/example_table.sqlx @@ -0,0 +1,5 @@ +config { + type: "table" +} + +select * from ${ref("sample_data")} as data diff --git a/tests/integration/mysql_project/definitions/example_view.sqlx b/tests/integration/mysql_project/definitions/example_view.sqlx new file mode 100644 index 00000000..6647979f --- /dev/null +++ b/tests/integration/mysql_project/definitions/example_view.sqlx @@ -0,0 +1,5 @@ +config { + type: "view" +} + +select * from ${ref("sample_data")} diff --git a/tests/integration/mysql_project/definitions/sample_data.sqlx b/tests/integration/mysql_project/definitions/sample_data.sqlx new file mode 100644 index 00000000..8ccb7308 --- /dev/null +++ b/tests/integration/mysql_project/definitions/sample_data.sqlx @@ -0,0 +1,7 @@ +config { + type: "view" +} + +select ${when(sqlanvil.projectConfig.vars.fooVar === "bar", "1", "2")} as val union all +select 2 as val union all +select ${when(sqlanvil.projectConfig.warehouse === "mysql", "3", "2")} as val diff --git a/tests/integration/mysql_project/definitions/sample_data_2.sqlx b/tests/integration/mysql_project/definitions/sample_data_2.sqlx new file mode 100644 index 00000000..4edd5e21 --- /dev/null +++ b/tests/integration/mysql_project/definitions/sample_data_2.sqlx @@ -0,0 +1,7 @@ +config { + type: "view" +} + +select 1 as val1, 1 as val2 union all +select 1 as val1, 2 as val2 union all +select 1 as val1, 3 as val2 diff --git a/tests/integration/mysql_project/package.json b/tests/integration/mysql_project/package.json new file mode 100644 index 00000000..0967ef42 --- /dev/null +++ b/tests/integration/mysql_project/package.json @@ -0,0 +1 @@ +{} diff --git a/tests/integration/mysql_project/workflow_settings.yaml b/tests/integration/mysql_project/workflow_settings.yaml new file mode 100644 index 00000000..e5bf8f61 --- /dev/null +++ b/tests/integration/mysql_project/workflow_settings.yaml @@ -0,0 +1,5 @@ +warehouse: mysql +defaultDataset: sa_integration_test +defaultAssertionDataset: sa_integration_test_assertions +vars: + fooVar: bar From 7666ca1573af8adfe43ff7d07db5d51f95737f6d Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 11:04:29 -0500 Subject: [PATCH 10/12] build(mysql): declare mysql2 as a bundle + package external Adds mysql2 to the cli package.json deps and mysql2/promise to the rollup externals (rollup matches the import specifier exactly), so the adapter's mysql2/promise import is not bundled. Fixes the cli bundle build. Co-Authored-By: Claude Opus 4.8 --- packages/@sqlanvil/cli/BUILD | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/packages/@sqlanvil/cli/BUILD b/packages/@sqlanvil/cli/BUILD index 9feb36e9..3088bebf 100644 --- a/packages/@sqlanvil/cli/BUILD +++ b/packages/@sqlanvil/cli/BUILD @@ -38,6 +38,7 @@ externals = [ "js-beautify", "js-yaml", "moo", + "mysql2", "object-sizeof", "parse-duration", "pg", @@ -54,6 +55,11 @@ externals = [ "yargs", ] +# Rollup matches import specifiers exactly, so the mysql2 promise subpath must be +# listed too (the adapter imports "mysql2/promise"); package.json only carries the +# real package name "mysql2". +bundle_externals = externals + ["mysql2/promise"] + pkg_json( name = "json", package_name = "@sqlanvil/cli", @@ -73,7 +79,7 @@ pkg_bundle( allow_node_builtins = True, args = ["--banner='#!/usr/bin/env node\n" + LICENSE_HEADER + "'"], entry_point = "index.ts", - externals = externals, + externals = bundle_externals, deps = [ ":cli", ], @@ -84,7 +90,7 @@ pkg_bundle( allow_node_builtins = True, args = ["--banner='#!/usr/bin/env node\n" + LICENSE_HEADER + "'"], entry_point = "worker.ts", - externals = externals, + externals = bundle_externals, deps = [ ":cli", ], From a1d26963d5a7efb60b79931bb9e888250bb130a7 Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 11:07:18 -0500 Subject: [PATCH 11/12] docs: add MySQL/MariaDB to README warehouse list Co-Authored-By: Claude Opus 4.8 --- README.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index d770d9a6..3b8b79de 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,8 @@ # SQLAnvil -**SQL workflow tool for BigQuery, Postgres, and Supabase.** +**SQL workflow tool for BigQuery, Postgres, Supabase, and MySQL/MariaDB.** -SQLAnvil is an open-source fork of [Dataform OSS](https://github.com/dataform-co/dataform) (Apache 2.0), extended with first-class PostgreSQL and Supabase support. Define your data transformations in SQLX, have SQLAnvil compile them to idiomatic SQL, and run them against your warehouse. +SQLAnvil is an open-source fork of [Dataform OSS](https://github.com/dataform-co/dataform) (Apache 2.0), extended with first-class PostgreSQL, Supabase, and MySQL/MariaDB support. Define your data transformations in SQLX, have SQLAnvil compile them to idiomatic SQL, and run them against your warehouse. > **SQLAnvil is not affiliated with or endorsed by Google.** The Dataform name and related marks are trademarks of Google LLC. See [NOTICE](NOTICE) for attribution. @@ -13,6 +13,7 @@ SQLAnvil is an open-source fork of [Dataform OSS](https://github.com/dataform-co - **BigQuery** — full support: partitioning, clustering, labels, materialized views, `MERGE`-based incremental upserts - **PostgreSQL** — idiomatic DDL: native partitioning, `INSERT ... ON CONFLICT` upserts, btree/gin/gist/brin indexes, tablespaces, fillfactor - **Supabase** — extends Postgres with RLS policies, Realtime publications, pgvector indexes, and Supabase Wrappers _(coming soon)_ +- **MySQL / MariaDB** — portable MySQL DDL: CTAS tables, `CREATE OR REPLACE VIEW`, `ON DUPLICATE KEY UPDATE` incremental upserts (one adapter, validated against both engines) - **SQLX + YAML + JS** — three authoring modes: SQL with config blocks, `actions.yaml` bulk definitions, or the JavaScript API --- From 12b3a00ee323a1267d8c0756bfd29da14bf3905a Mon Sep 17 00:00:00 2001 From: Ivan Histand Date: Wed, 17 Jun 2026 11:27:56 -0500 Subject: [PATCH 12/12] =?UTF-8?q?refactor(mysql):=20address=20review=20?= =?UTF-8?q?=E2=80=94=20convertFieldType=20coverage=20+=20invariant=20comme?= =?UTF-8?q?nts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - convertFieldType maps BIT->INTEGER, JSON/TIME->STRING; documents binary/blob/ geometry falling through to UNKNOWN (introspection-only metadata) - comment the upsertInto no-metadata fallback invariant and the byteLimit/ client-side rowLimit divergence from the streaming Postgres path Co-Authored-By: Claude Opus 4.8 --- cli/api/dbadapters/mysql_execution_sql.ts | 5 +++++ cli/api/utils/mysql.ts | 8 ++++++++ 2 files changed, 13 insertions(+) diff --git a/cli/api/dbadapters/mysql_execution_sql.ts b/cli/api/dbadapters/mysql_execution_sql.ts index 1f645196..ed79dc45 100644 --- a/cli/api/dbadapters/mysql_execution_sql.ts +++ b/cli/api/dbadapters/mysql_execution_sql.ts @@ -113,6 +113,11 @@ export class MysqlExecutionSql implements IExecutionSql { const columns = (tableMetadata?.fields || []).map(f => f.name); const query = this.getIncrementalQuery(table); if (columns.length === 0) { + // No known columns means no tableMetadata — but this path only runs when + // shouldWriteIncrementally() already required tableMetadata, so columns are + // populated in practice. The bare insert is a defensive fallback; with a + // uniqueKey present it could hit the unique index, so it's intentionally + // never the upsert path. return `insert into ${target} select * from (${query}) as insertions`; } const backticked = columns.map(c => `\`${c}\``); diff --git a/cli/api/utils/mysql.ts b/cli/api/utils/mysql.ts index d6d6abcb..9bd03ad5 100644 --- a/cli/api/utils/mysql.ts +++ b/cli/api/utils/mysql.ts @@ -60,6 +60,9 @@ export class MySqlPoolExecutor { ): Promise => { const [rows] = await conn.query(statement, options.params || []); const arr = Array.isArray(rows) ? (rows as any[]) : []; + // mysql2 buffers the full result set (no streaming cursor), so rowLimit + // is applied client-side after the fetch. byteLimit is not enforced here + // — unlike the Postgres adapter's streaming LimitedResultSet. return options.rowLimit && arr.length > options.rowLimit ? arr.slice(0, options.rowLimit) : arr; @@ -91,6 +94,7 @@ export function convertFieldType(type: string) { case "INTEGER": case "BIGINT": case "YEAR": + case "BIT": return sqlanvil.Field.Primitive.INTEGER; case "DECIMAL": case "DEC": @@ -108,12 +112,16 @@ export function convertFieldType(type: string) { case "LONGTEXT": case "ENUM": case "SET": + case "JSON": + case "TIME": return sqlanvil.Field.Primitive.STRING; case "DATE": return sqlanvil.Field.Primitive.DATE; case "DATETIME": case "TIMESTAMP": return sqlanvil.Field.Primitive.TIMESTAMP; + // BINARY/VARBINARY, the BLOB family, and GEOMETRY have no field primitive; + // they fall through to UNKNOWN (introspection metadata only). default: return sqlanvil.Field.Primitive.UNKNOWN; }