diff --git a/.changeset/pull-clickhouse.md b/.changeset/pull-clickhouse.md new file mode 100644 index 0000000..de79007 --- /dev/null +++ b/.changeset/pull-clickhouse.md @@ -0,0 +1,11 @@ +--- +"@maple-dev/effect-orm": minor +--- + +`effect-orm pull` writes table definitions from an existing ClickHouse database, so adopting effect-orm on a database that already has tables no longer means writing every `CH.table` by hand. + +- It reads the database the config's `driver` connects to and writes one `CH.table` per table: engine and arguments, keys as declared, partition key, TTL, settings, comments, column defaults, codecs and skipping indexes. The definitions yield exactly what the database holds, so `generate --baseline` adopts them and the next `generate` has nothing to do. +- Column types map to the built-ins that write the same SQL; others become `CH.custom` with a fitting schema (`UUID`, `Date`, `Decimal`, `Enum8`, …) or `CH.untyped`. Views and non-MergeTree tables are written as `external: true`. Materialized views, `SAMPLE BY` and `EPHEMERAL` columns are listed for review in the file's header. +- `pull` is also exported from `/kit`. + +`verify` reads a table's sorting and primary keys as declared. It had compared `system.tables.sorting_key`, which has a `VersionedCollapsingMergeTree`'s version column appended, so such a table always showed drift. diff --git a/.gitignore b/.gitignore index 96994e2..b5b42c6 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ coverage/ *.tgz .env* .DS_Store +tests/.pull-* diff --git a/README.md b/README.md index 20e0b25..4863efe 100644 --- a/README.md +++ b/README.md @@ -226,7 +226,7 @@ regressions live in [`src/docs-examples.test.ts`](./src/docs-examples.test.ts). | `@maple-dev/effect-orm/sql` | The low-level `SqlFragment` AST (`raw`, `ident`, `compile`, …) for hand-rolling fragments. | | `@maple-dev/effect-orm/tinybird` | Tinybird datasources and materialized views, SDK-compatible, that are also query tables; `buildProject` writes the datafiles. See [Tinybird](./docs/tinybird.md). | | `@maple-dev/effect-orm/schema` | Migration tooling over `table` values: DDL rendering, snapshots, the schema diff. Pure. | -| `@maple-dev/effect-orm/kit`, `/migrate` | `effect-orm generate` and `check`; applying migrations through a driver you provide. See [Schema and migrations](./docs/migrations.md). | +| `@maple-dev/effect-orm/kit`, `/migrate` | `effect-orm generate`, `check` and `pull` (definitions from an existing ClickHouse database); applying migrations through a driver you provide. See [Schema and migrations](./docs/migrations.md). | | `@maple-dev/effect-orm/database` | `Database` over your `SqlClient`: `run`, `execute`, `transaction` with retry. | | `@maple-dev/effect-orm/clickhouse-http` | A ClickHouse `SqlClient` over `fetch`, for Cloudflare Workers and any runtime with `fetch`; give it to `Database`. | diff --git a/docs/migrations.md b/docs/migrations.md index 6d731fc..67f9e26 100644 --- a/docs/migrations.md +++ b/docs/migrations.md @@ -151,6 +151,45 @@ tables. Independent branches are fine: the next `generate` records both as paren change the same table conflict; delete one migration and generate it again on top of the other. Run `check` in CI. +### Starting from an existing database + +A ClickHouse database that predates effect-orm does not need its tables written by hand. +`effect-orm pull` reads the database the config's `driver` connects to and writes a schema +module, one `CH.table` per table; then `generate --baseline` starts the history from it: + +```sh +effect-orm pull # writes the config's schema module; --out for another +effect-orm generate --baseline # a migration that runs nothing, with this schema as its snapshot +effect-orm migrate # runs no statement; records the baseline as applied +``` + +The module is written so that its definitions yield exactly what the database holds: engine +and its arguments, keys as declared, partition key, TTL, settings, comments, column defaults +(`DEFAULT`, `MATERIALIZED`, `ALIAS`), codecs and skipping indexes, with every expression as the +server's own text. The first `generate` after the baseline finds nothing to do. A `Replicated` +or `Shared` engine is written as its plain family; replication is a render option. + +Column types map to the built-ins that write the same SQL (`LowCardinality(Nullable(String))` is +`CH.lowCardinality(CH.nullable(CH.string))`). A type without one becomes `CH.custom` with the +schema its JSON value decodes with (`UUID` and `Date` as strings, `Decimal` and `Int8` as +numbers, an `Enum8` as its labels), and a type the generator does not know (`Tuple`, `JSON`, a +`Map` keyed by numbers) becomes `CH.untyped`. The file's header lists what to review: + +- **Materialized views** are not generated: a view's body is a query-DSL value, which SQL text + cannot be turned back into. The header quotes each body; write it with `CH.materializedView`. +- **Views, dictionaries, `Distributed`, `Kafka` and other non-MergeTree tables** are written as + `external: true`: queryable, never migrated. +- A table with a name that is not a plain identifier (`my-col`), or a MergeTree declared in the + old `MergeTree(date, key, granularity)` syntax, is written as external too. +- **Not modeled, so left out:** `SAMPLE BY`, projections, constraints and `EPHEMERAL` columns. + Column-level TTLs are not read at all. +- A `Decimal` wider than 15 digits decodes to a number, which rounds; the header flags each one. +- A user who may not read `system.data_skipping_indices` gets no indexes, and a note saying so. + +`--tables a,b` writes only those tables; `--force` overwrites an existing file. `verify` reports +the views and materialized views as `unexpected` until the schema declares them. Postgres is +not supported yet. + ## Applying migrations The library opens no connection. Give it a `MigrationDriver`, usually built from the diff --git a/docs/reference.md b/docs/reference.md index 89c2faf..abc2d46 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -515,7 +515,7 @@ placeholders, double-quoted identifiers. | `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` | | `withQuerySummary`, `QuerySummary` — the `db.query.summary` of the statements an effect runs, for spans | `/database` | | `ClickhouseHttpClient`, `layer`, `make`, `ClickhouseHttpClientConfig` — a ClickHouse `SqlClient` over `fetch`, for Workers | `/clickhouse-http` | -| `defineConfig`, `generate`, `check`, `loadSchema`, `readMigrations`, `analyze`, `KitError` | `/kit` | +| `defineConfig`, `generate`, `check`, `pull`, `loadSchema`, `readMigrations`, `analyze`, `KitError` | `/kit` | | `defineSuite`, `query`, `caseFromCompiled`, `runSuite`, `compareRuns`, `compareBudgets` | `/benchmark` | | `Suite`, `RunOutput`, `BenchmarkError` and benchmark contracts | `/benchmark` | | `makeHttpClient`, `makeHttpTransport`, `httpConfigFromEnv` | `/benchmark/http` | diff --git a/src/kit.ts b/src/kit.ts index d3b23cd..5ffef5f 100644 --- a/src/kit.ts +++ b/src/kit.ts @@ -16,4 +16,5 @@ export { type GenerateResult, type KitConfig, } from "./kit/generate" +export { pull, type PullOptions, type PullResult } from "./kit/pull" export { runCli } from "./kit/cli" diff --git a/src/kit/__fixtures__/clickhouse-catalog.json b/src/kit/__fixtures__/clickhouse-catalog.json new file mode 100644 index 0000000..59120e2 --- /dev/null +++ b/src/kit/__fixtures__/clickhouse-catalog.json @@ -0,0 +1,717 @@ +{ + "database": "probe", + "tables": [ + { + "name": "_effect_orm_migrations", + "engine": "MergeTree", + "engine_full": "MergeTree ORDER BY name SETTINGS index_granularity = 8192", + "sorting_key": "name", + "partition_key": "", + "primary_key": "name", + "sampling_key": "", + "as_select": "", + "comment": "" + }, + { + "name": "coll", + "engine": "VersionedCollapsingMergeTree", + "engine_full": "VersionedCollapsingMergeTree(Sign, Ver) ORDER BY K SETTINGS index_granularity = 8192", + "sorting_key": "K, Ver", + "partition_key": "", + "primary_key": "K", + "sampling_key": "", + "as_select": "", + "comment": "" + }, + { + "name": "dist", + "engine": "Distributed", + "engine_full": "Distributed('default', 'probe', 't')", + "sorting_key": "", + "partition_key": "", + "primary_key": "", + "sampling_key": "", + "as_select": "", + "comment": "events table" + }, + { + "name": "mem", + "engine": "Memory", + "engine_full": "Memory", + "sorting_key": "", + "partition_key": "", + "primary_key": "", + "sampling_key": "", + "as_select": "", + "comment": "" + }, + { + "name": "notuple", + "engine": "MergeTree", + "engine_full": "MergeTree ORDER BY tuple() SETTINGS index_granularity = 8192", + "sorting_key": "", + "partition_key": "", + "primary_key": "", + "sampling_key": "", + "as_select": "", + "comment": "" + }, + { + "name": "plain", + "engine": "MergeTree", + "engine_full": "MergeTree ORDER BY (`my-col`, Id) SETTINGS index_granularity = 8192", + "sorting_key": "`my-col`, Id", + "partition_key": "", + "primary_key": "`my-col`, Id", + "sampling_key": "", + "as_select": "", + "comment": "" + }, + { + "name": "sink", + "engine": "Null", + "engine_full": "`Null`", + "sorting_key": "", + "partition_key": "", + "primary_key": "", + "sampling_key": "", + "as_select": "", + "comment": "" + }, + { + "name": "t", + "engine": "ReplacingMergeTree", + "engine_full": "ReplacingMergeTree(Ts) PARTITION BY toDate(Ts) PRIMARY KEY (OrgId, H) ORDER BY (OrgId, H, toStartOfHour(Ts), Name) SAMPLE BY H TTL toDate(Ts) + toIntervalDay(30) SETTINGS index_granularity = 4096, storage_policy = 'default'", + "sorting_key": "OrgId, H, toStartOfHour(Ts), Name", + "partition_key": "toDate(Ts)", + "primary_key": "OrgId, H", + "sampling_key": "H", + "as_select": "", + "comment": "events table" + }, + { + "name": "totals", + "engine": "SummingMergeTree", + "engine_full": "SummingMergeTree ORDER BY (OrgId, Name) SETTINGS index_granularity = 8192", + "sorting_key": "OrgId, Name", + "partition_key": "", + "primary_key": "OrgId, Name", + "sampling_key": "", + "as_select": "", + "comment": "" + }, + { + "name": "totals_mv", + "engine": "MaterializedView", + "engine_full": "", + "sorting_key": "", + "partition_key": "", + "primary_key": "", + "sampling_key": "", + "as_select": "SELECT OrgId, Name, count() AS C FROM probe.t GROUP BY OrgId, Name", + "comment": "" + }, + { + "name": "v", + "engine": "View", + "engine_full": "", + "sorting_key": "", + "partition_key": "", + "primary_key": "", + "sampling_key": "", + "as_select": "SELECT OrgId FROM probe.t", + "comment": "" + } + ], + "columns": [ + { + "table": "_effect_orm_migrations", + "name": "name", + "type": "String", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + }, + { + "table": "coll", + "name": "K", + "type": "String", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + }, + { + "table": "coll", + "name": "Sign", + "type": "Int8", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 2 + }, + { + "table": "coll", + "name": "Ver", + "type": "UInt32", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 3 + }, + { + "table": "dist", + "name": "OrgId", + "type": "LowCardinality(String)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + }, + { + "table": "dist", + "name": "Ts", + "type": "DateTime64(3)", + "default_kind": "", + "default_expression": "", + "compression_codec": "CODEC(Delta(8), ZSTD(1))", + "comment": "", + "position": 2 + }, + { + "table": "dist", + "name": "Name", + "type": "String", + "default_kind": "DEFAULT", + "default_expression": "'x'", + "compression_codec": "", + "comment": "the name", + "position": 3 + }, + { + "table": "dist", + "name": "N", + "type": "Nullable(UInt64)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 4 + }, + { + "table": "dist", + "name": "M", + "type": "Map(String, Array(Float64))", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 5 + }, + { + "table": "dist", + "name": "E", + "type": "Enum8('a' = 1, 'b' = 2)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 6 + }, + { + "table": "dist", + "name": "D", + "type": "Decimal(18, 4)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 7 + }, + { + "table": "dist", + "name": "U", + "type": "UUID", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 8 + }, + { + "table": "dist", + "name": "V", + "type": "UInt8", + "default_kind": "MATERIALIZED", + "default_expression": "1", + "compression_codec": "", + "comment": "", + "position": 9 + }, + { + "table": "dist", + "name": "A", + "type": "AggregateFunction(quantiles(0.5, 0.9), UInt64)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 10 + }, + { + "table": "dist", + "name": "S", + "type": "SimpleAggregateFunction(sum, UInt64)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 11 + }, + { + "table": "dist", + "name": "Z", + "type": "DateTime('UTC')", + "default_kind": "ALIAS", + "default_expression": "Ts", + "compression_codec": "", + "comment": "", + "position": 12 + }, + { + "table": "dist", + "name": "LN", + "type": "LowCardinality(Nullable(String))", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 13 + }, + { + "table": "dist", + "name": "H", + "type": "UInt64", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 14 + }, + { + "table": "mem", + "name": "A", + "type": "String", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + }, + { + "table": "mem", + "name": "N", + "type": "UInt32", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 2 + }, + { + "table": "notuple", + "name": "A", + "type": "String", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + }, + { + "table": "plain", + "name": "my-col", + "type": "String", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + }, + { + "table": "plain", + "name": "Id", + "type": "UInt64", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 2 + }, + { + "table": "plain", + "name": "B", + "type": "Map(UInt64, String)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 3 + }, + { + "table": "plain", + "name": "T", + "type": "Tuple(a String, b UInt8)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 4 + }, + { + "table": "plain", + "name": "J", + "type": "JSON", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 5 + }, + { + "table": "plain", + "name": "Tz", + "type": "DateTime64(3, 'UTC')", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 6 + }, + { + "table": "plain", + "name": "NT", + "type": "Nullable(DateTime64(3))", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 7 + }, + { + "table": "plain", + "name": "F", + "type": "FixedString(4)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 8 + }, + { + "table": "plain", + "name": "Day", + "type": "Date", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 9 + }, + { + "table": "plain", + "name": "I8", + "type": "Int8", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 10 + }, + { + "table": "plain", + "name": "F32", + "type": "Float32", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 11 + }, + { + "table": "plain", + "name": "Big", + "type": "Int128", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 12 + }, + { + "table": "plain", + "name": "Ip", + "type": "IPv4", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 13 + }, + { + "table": "plain", + "name": "Q", + "type": "Enum8('it\\'s' = 1, 'b' = 2)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 14 + }, + { + "table": "plain", + "name": "Eph", + "type": "UInt8", + "default_kind": "EPHEMERAL", + "default_expression": "defaultValueOfTypeName('UInt8')", + "compression_codec": "", + "comment": "", + "position": 15 + }, + { + "table": "sink", + "name": "A", + "type": "String", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + }, + { + "table": "t", + "name": "OrgId", + "type": "LowCardinality(String)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + }, + { + "table": "t", + "name": "Ts", + "type": "DateTime64(3)", + "default_kind": "", + "default_expression": "", + "compression_codec": "CODEC(Delta(8), ZSTD(1))", + "comment": "", + "position": 2 + }, + { + "table": "t", + "name": "Name", + "type": "String", + "default_kind": "DEFAULT", + "default_expression": "'x'", + "compression_codec": "", + "comment": "the name", + "position": 3 + }, + { + "table": "t", + "name": "N", + "type": "Nullable(UInt64)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 4 + }, + { + "table": "t", + "name": "M", + "type": "Map(String, Array(Float64))", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 5 + }, + { + "table": "t", + "name": "E", + "type": "Enum8('a' = 1, 'b' = 2)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 6 + }, + { + "table": "t", + "name": "D", + "type": "Decimal(18, 4)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 7 + }, + { + "table": "t", + "name": "U", + "type": "UUID", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 8 + }, + { + "table": "t", + "name": "V", + "type": "UInt8", + "default_kind": "MATERIALIZED", + "default_expression": "1", + "compression_codec": "", + "comment": "", + "position": 9 + }, + { + "table": "t", + "name": "A", + "type": "AggregateFunction(quantiles(0.5, 0.9), UInt64)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 10 + }, + { + "table": "t", + "name": "S", + "type": "SimpleAggregateFunction(sum, UInt64)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 11 + }, + { + "table": "t", + "name": "Z", + "type": "DateTime('UTC')", + "default_kind": "ALIAS", + "default_expression": "Ts", + "compression_codec": "", + "comment": "", + "position": 12 + }, + { + "table": "t", + "name": "LN", + "type": "LowCardinality(Nullable(String))", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 13 + }, + { + "table": "t", + "name": "H", + "type": "UInt64", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 14 + }, + { + "table": "totals", + "name": "OrgId", + "type": "String", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + }, + { + "table": "totals", + "name": "Name", + "type": "String", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 2 + }, + { + "table": "totals", + "name": "C", + "type": "UInt64", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 3 + }, + { + "table": "totals_mv", + "name": "OrgId", + "type": "LowCardinality(String)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + }, + { + "table": "totals_mv", + "name": "Name", + "type": "String", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 2 + }, + { + "table": "totals_mv", + "name": "C", + "type": "UInt64", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 3 + }, + { + "table": "v", + "name": "OrgId", + "type": "LowCardinality(String)", + "default_kind": "", + "default_expression": "", + "compression_codec": "", + "comment": "", + "position": 1 + } + ], + "indexes": [ + { + "table": "t", + "name": "idx_name", + "type_full": "bloom_filter(0.01)", + "expr": "Name", + "granularity": 2 + } + ] +} diff --git a/src/kit/__fixtures__/pulled-clickhouse.ts b/src/kit/__fixtures__/pulled-clickhouse.ts new file mode 100644 index 0000000..09f438a --- /dev/null +++ b/src/kit/__fixtures__/pulled-clickhouse.ts @@ -0,0 +1,137 @@ +// Generated by `effect-orm pull` from ClickHouse database `probe`. +// +// Review: +// - plain.Eph: EPHEMERAL columns are not modeled; left out +// - plain: generated as external; table definitions take plain identifiers only, not "my-col" +// - t: SAMPLE BY H is not modeled; generated without it +// - totals_mv: materialized view not generated; write it with CH.materializedView. Its body: SELECT OrgId, Name, count() AS C FROM t GROUP BY OrgId, Name +// - dist.D: Decimal(18, 4) decodes to a number, which rounds past 15 significant digits +// - plain.B: Map(UInt64, String) has no built-in type; rows hold it undecoded +// - plain.T: Tuple(a String, b UInt8) has no built-in type; rows hold it undecoded +// - plain.J: JSON has no built-in type; rows hold it undecoded +// - t.D: Decimal(18, 4) decodes to a number, which rounds past 15 significant digits + +import { Schema } from "effect" +import * as CH from "../../clickhouse" + +export const Coll = CH.table("coll", { + columns: { + K: CH.string, + Sign: CH.custom("Int8", CH.CHNumber), + Ver: CH.uint32, + }, + engine: CH.engine.versionedCollapsingMergeTree("Sign", "Ver"), + orderBy: ["K"], +}) + +/** Distributed: queried, never migrated. */ +export const Dist = CH.table("dist", { + external: true, + columns: { + OrgId: CH.lowCardinality(CH.string), + Ts: CH.precision(CH.dateTime64, 3), + Name: CH.column(CH.string, { defaultExpr: "'x'" }), + N: CH.nullable(CH.uint64), + M: CH.map(CH.string, CH.array(CH.float64)), + E: CH.custom("Enum8('a' = 1, 'b' = 2)", Schema.Literals(["a", "b"])), + D: CH.custom("Decimal(18, 4)", CH.CHNumber), + U: CH.custom("UUID", Schema.String), + V: CH.column(CH.uint8, { materialized: "1" }), + A: CH.aggregateState("quantiles(0.5, 0.9)", "UInt64"), + S: CH.simpleAggregateFunction("sum", CH.uint64), + Z: CH.column(CH.custom("DateTime('UTC')", Schema.String), { alias: "Ts" }), + LN: CH.lowCardinality(CH.nullable(CH.string)), + H: CH.uint64, + }, +}) + +export const Mem = CH.table("mem", { + columns: { + A: CH.string, + N: CH.uint32, + }, + engine: CH.engine.memory(), +}) + +export const Notuple = CH.table("notuple", { + columns: { + A: CH.string, + }, + engine: CH.engine.mergeTree(), + orderBy: [], +}) + +/** MergeTree: queried, never migrated. */ +export const Plain = CH.table("plain", { + external: true, + columns: { + "my-col": CH.string, + Id: CH.uint64, + B: CH.untyped("Map(UInt64, String)"), + T: CH.untyped("Tuple(a String, b UInt8)"), + J: CH.untyped("JSON"), + Tz: CH.custom("DateTime64(3, 'UTC')", Schema.String), + NT: CH.nullable(CH.precision(CH.dateTime64, 3)), + F: CH.custom("FixedString(4)", Schema.String), + Day: CH.custom("Date", Schema.String), + I8: CH.custom("Int8", CH.CHNumber), + F32: CH.custom("Float32", CH.CHNumber), + Big: CH.custom("Int128", Schema.String), + Ip: CH.custom("IPv4", Schema.String), + Q: CH.custom("Enum8('it\\'s' = 1, 'b' = 2)", Schema.Literals(["it's", "b"])), + }, +}) + +export const Sink = CH.table("sink", { + columns: { + A: CH.string, + }, + engine: CH.engine.null(), +}) + +export const T = CH.table("t", { + columns: { + OrgId: CH.lowCardinality(CH.string), + Ts: CH.column(CH.precision(CH.dateTime64, 3), { codec: "Delta(8), ZSTD(1)" }), + Name: CH.column(CH.string, { defaultExpr: "'x'", comment: "the name" }), + N: CH.nullable(CH.uint64), + M: CH.map(CH.string, CH.array(CH.float64)), + E: CH.custom("Enum8('a' = 1, 'b' = 2)", Schema.Literals(["a", "b"])), + D: CH.custom("Decimal(18, 4)", CH.CHNumber), + U: CH.custom("UUID", Schema.String), + V: CH.column(CH.uint8, { materialized: "1" }), + A: CH.aggregateState("quantiles(0.5, 0.9)", "UInt64"), + S: CH.simpleAggregateFunction("sum", CH.uint64), + Z: CH.column(CH.custom("DateTime('UTC')", Schema.String), { alias: "Ts" }), + LN: CH.lowCardinality(CH.nullable(CH.string)), + H: CH.uint64, + }, + engine: CH.engine.replacingMergeTree({ version: "Ts" }), + orderBy: () => ["OrgId", "H", "toStartOfHour(Ts)", "Name"], + partitionBy: "toDate(Ts)", + primaryKey: ["OrgId", "H"], + ttl: "toDate(Ts) + toIntervalDay(30)", + settings: { index_granularity: "4096", storage_policy: "'default'" }, + indexes: [ + CH.index("idx_name", "Name", "bloom_filter(0.01)", 2), + ], + comment: "events table", +}) + +export const Totals = CH.table("totals", { + columns: { + OrgId: CH.string, + Name: CH.string, + C: CH.uint64, + }, + engine: CH.engine.summingMergeTree(), + orderBy: ["OrgId", "Name"], +}) + +/** View: queried, never migrated. */ +export const V = CH.table("v", { + external: true, + columns: { + OrgId: CH.lowCardinality(CH.string), + }, +}) diff --git a/src/kit/cli.ts b/src/kit/cli.ts index cb856bd..9cbf026 100644 --- a/src/kit/cli.ts +++ b/src/kit/cli.ts @@ -7,6 +7,7 @@ import { readFile } from "node:fs/promises" import * as Migrate from "../migrate" import { Hints, type Hint } from "../schema/diff" import { check, generate, KitError, readMigrations, type KitConfig } from "./generate" +import { pull } from "./pull" const help = `effect-orm: schema migrations for ClickHouse and Postgres @@ -14,6 +15,9 @@ const help = `effect-orm: schema migrations for ClickHouse and Postgres [--hints ] [--hints-file ] [--baseline [--from-drizzle]] Start the history of a folder another tool wrote: a migration that runs nothing, with the current schema (or drizzle-kit's last snapshot) as its snapshot + pull [--out ] [--tables a,b] [--force] + Write table definitions from the database (ClickHouse). Default --out: the + config's schema module, when it names one. Then run generate --baseline check Validate the migrations folder (snapshot chain, branch conflicts) migrate [--strict] Apply pending migrations (needs config.driver) status Applied, pending, partial, or changed, per migration @@ -36,6 +40,9 @@ const options = { hints: { type: "string" }, "hints-file": { type: "string" }, strict: { type: "boolean" }, + out: { type: "string" }, + tables: { type: "string" }, + force: { type: "boolean" }, ran: { type: "boolean" }, "not-ran": { type: "boolean" }, json: { type: "boolean" }, @@ -114,6 +121,28 @@ const program = (args: ReadonlyArray, print: (line: string) => void) => ) return 0 } + case "pull": { + const modules = typeof config.schema === "string" ? [config.schema] : config.schema + const target = parsed.values.out ?? (modules.length === 1 ? modules[0] : undefined) + if (target === undefined) { + return yield* new KitError({ code: "config", message: "the config names several schema modules; say which to write with --out" }) + } + const tables = parsed.values.tables?.split(",").map((t) => t.trim()).filter((t) => t.length > 0) + if (tables !== undefined && tables.length === 0) { + return yield* new KitError({ code: "config", message: "--tables takes a comma-separated list of table names" }) + } + const result = yield* pull(config, cwd, { + out: target, + ...(tables !== undefined ? { tables } : undefined), + ...(parsed.values.force === true ? { force: true } : undefined), + }) + out(result, () => [ + `Wrote ${result.written}: ${result.tables.length} tables${result.external.length > 0 ? `, ${result.external.length} external (${result.external.join(", ")})` : ""}.`, + ...(result.review.length > 0 ? ["Review:", ...result.review.map((line) => ` ${line}`)] : []), + "Next: effect-orm generate --baseline, to start the migration history from this schema.", + ]) + return 0 + } case "check": { const result = yield* check(config, cwd) out(result, () => [ diff --git a/src/kit/codegen.ts b/src/kit/codegen.ts new file mode 100644 index 0000000..47958ed --- /dev/null +++ b/src/kit/codegen.ts @@ -0,0 +1,295 @@ +// `effect-orm pull`'s output: a schema module written from a live database. +// +// Every value is written so that the definition yields back the entity it +// came from: `entitiesOf` on the generated module equals the introspected +// entities. Expressions (defaults, keys, TTL, index bodies) are the server's +// text as strings; turning SQL back into the query DSL is not attempted. +// +// A column type maps to the built-in that writes the same SQL (`CH.nullable`, +// `CH.precision(CH.dateTime64, 3)`, ...). A type with no built-in becomes +// `CH.custom` with the schema its JSON value decodes with, and one this file +// does not know becomes `CH.untyped`, marked for review. + +import type { ColumnEntity, EngineSpec, IndexEntity, TableEntity } from "../schema/entities" +import { keyParts, splitCall, type ExternalTable, type IntrospectedSchema } from "../migrate/introspect" + +export interface CodegenOptions { + /** Where the module imports `CH` from. Default `@maple-dev/effect-orm/clickhouse`. */ + readonly importPath?: string + /** Lines for the header comment, after the first. */ + readonly header?: ReadonlyArray +} + +const str = (value: string): string => JSON.stringify(value) +const IDENTIFIER = /^[A-Za-z_$][A-Za-z0-9_$]*$/ +const key = (name: string): string => (IDENTIFIER.test(name) ? name : str(name)) + +// Column types + +interface TypeCode { + readonly code: string + /** The SQL the code writes; a mismatch with the column's type falls back to `CH.custom`. */ + readonly sql: string + /** It decodes to a string, so it can key a `Map`. */ + readonly stringKey: boolean + readonly usesSchema: boolean + readonly todo?: string +} + +const SCALARS: Readonly> = { + String: "CH.string", + UInt8: "CH.uint8", + UInt16: "CH.uint16", + UInt32: "CH.uint32", + UInt64: "CH.uint64", + Int32: "CH.int32", + Int64: "CH.int64", + Float64: "CH.float64", + Bool: "CH.bool", + DateTime: "CH.dateTime", +} +/** Read as `number`: what `CHNumber` accepts on the wire (a number, or a quoted one). */ +const NUMBERS = new Set(["Int8", "Int16", "Float32", "BFloat16", "Decimal", "Decimal32", "Decimal64", "Decimal128", "Decimal256"]) +/** Read as `string`: values JSON quotes, or a 128/256-bit integer `number` would round. */ +const STRINGS = new Set(["UInt128", "UInt256", "Int128", "Int256", "Date", "Date32", "UUID", "IPv4", "IPv6", "FixedString", "DateTime", "DateTime64"]) + +const custom = (sql: string, schema: string, stringKey: boolean): TypeCode => ({ + code: `CH.custom(${str(sql)}, ${schema})`, + sql, + stringKey, + usesSchema: schema.startsWith("Schema."), +}) + +const untyped = (sql: string): TypeCode => ({ + code: `CH.untyped(${str(sql)})`, + sql, + stringKey: false, + usesSchema: false, + todo: `${sql} has no built-in type; rows hold it undecoded`, +}) + +/** `'a' = 1` as `a`. */ +const enumLabel = (entry: string): string | undefined => { + const match = /^'((?:[^'\\]|\\.)*)'\s*=\s*-?\d+$/s.exec(entry.trim()) + return match === null ? undefined : match[1]!.replace(/\\(.)/g, "$1") +} + +const mapped = (sql: string): TypeCode => { + const { name, args } = splitCall(sql) + if (args === undefined) { + const scalar = SCALARS[name] + if (scalar !== undefined) return { code: scalar, sql, stringKey: name === "String", usesSchema: false } + if (NUMBERS.has(name)) return custom(sql, "CH.CHNumber", false) + if (STRINGS.has(name)) return custom(sql, "Schema.String", true) + return untyped(sql) + } + /** `open` is the code before the inner type's: `CH.nullable(`. */ + const wrap = (open: string, inner: TypeCode, sqlOf: (s: string) => string, stringKey = false): TypeCode => ({ + code: `${open}${inner.code})`, + sql: sqlOf(inner.sql), + stringKey, + usesSchema: inner.usesSchema, + ...(inner.todo !== undefined ? { todo: inner.todo } : undefined), + }) + switch (name) { + case "Nullable": + return args.length === 1 ? wrap("CH.nullable(", typeCode(args[0]!), (s) => `Nullable(${s})`) : untyped(sql) + case "LowCardinality": { + if (args.length !== 1) return untyped(sql) + const inner = typeCode(args[0]!) + return wrap("CH.lowCardinality(", inner, (s) => `LowCardinality(${s})`, inner.stringKey) + } + case "Array": + return args.length === 1 ? wrap("CH.array(", typeCode(args[0]!), (s) => `Array(${s})`) : untyped(sql) + case "Map": { + if (args.length !== 2) return untyped(sql) + const [k, v] = [typeCode(args[0]!), typeCode(args[1]!)] + if (!k.stringKey) return untyped(sql) + const todo = k.todo ?? v.todo + return { + code: `CH.map(${k.code}, ${v.code})`, + sql: `Map(${k.sql}, ${v.sql})`, + stringKey: false, + usesSchema: k.usesSchema || v.usesSchema, + ...(todo !== undefined ? { todo } : undefined), + } + } + case "DateTime64": + return args.length === 1 && /^\d$/.test(args[0]!) + ? { code: `CH.precision(CH.dateTime64, ${args[0]})`, sql: `DateTime64(${args[0]})`, stringKey: false, usesSchema: false } + : custom(sql, "Schema.String", true) + case "SimpleAggregateFunction": + if (args.length !== 2) return untyped(sql) + return wrap(`CH.simpleAggregateFunction(${str(args[0]!)}, `, typeCode(args[1]!), (s) => `SimpleAggregateFunction(${args[0]}, ${s})`) + case "AggregateFunction": + return { + code: `CH.aggregateState(${args.map(str).join(", ")})`, + sql: `AggregateFunction(${args.join(", ")})`, + stringKey: false, + usesSchema: false, + } + case "Enum8": + case "Enum16": { + const labels = args.map(enumLabel) + if (labels.length === 0 || labels.some((label) => label === undefined)) return untyped(sql) + return custom(sql, `Schema.Literals([${labels.map((label) => str(label!)).join(", ")}])`, true) + } + case "Decimal": + case "Decimal32": + case "Decimal64": + case "Decimal128": + case "Decimal256": { + const precision = name === "Decimal" ? Number(args[0]) : { Decimal32: 9, Decimal64: 18, Decimal128: 38, Decimal256: 76 }[name] + const type = custom(sql, "CH.CHNumber", false) + // A double holds 15 significant digits exactly. + return precision > 15 ? { ...type, todo: `${sql} decodes to a number, which rounds past 15 significant digits` } : type + } + default: + if (NUMBERS.has(name)) return custom(sql, "CH.CHNumber", false) + if (STRINGS.has(name)) return custom(sql, "Schema.String", true) + return untyped(sql) + } +} + +/** The code for a column type, writing exactly `sql`. */ +export const typeCode = (sql: string): TypeCode => { + const result = mapped(sql) + if (result.sql === sql) return result + // A built-in that writes the type differently: keep the server's text. + return untyped(sql) +} + +// Tables + +interface Out { + readonly lines: Array + usesSchema: boolean + readonly todos: Array +} + +/** `ddl: false` writes only the default's kind, which says what an insert may leave out or not write. */ +const columnCode = (out: Out, column: ColumnEntity, ddl: boolean): string => { + const type = typeCode(column.type) + if (type.usesSchema) out.usesSchema = true + if (type.todo !== undefined) out.todos.push(`${column.table}.${column.name}: ${type.todo}`) + const options: Array = [] + if (column.default !== null) { + const option = { DEFAULT: "defaultExpr", MATERIALIZED: "materialized", ALIAS: "alias" }[column.default.kind] + options.push(`${option}: ${str(column.default.expr)}`) + } + if (ddl) { + if (column.codec !== null) options.push(`codec: ${str(column.codec)}`) + if (column.comment !== null) options.push(`comment: ${str(column.comment)}`) + } + return options.length === 0 ? type.code : `CH.column(${type.code}, { ${options.join(", ")} })` +} + +const engineCode = (engine: EngineSpec): string => { + const [a, b] = engine.params + const n = engine.params.length + const helpers: Readonly> = { + MergeTree: "mergeTree", + AggregatingMergeTree: "aggregatingMergeTree", + ReplacingMergeTree: "replacingMergeTree", + SummingMergeTree: "summingMergeTree", + Null: "null", + Memory: "memory", + } + if (n === 0 && helpers[engine.family] !== undefined) return `CH.engine.${helpers[engine.family]}()` + if (engine.family === "ReplacingMergeTree" && n <= 2) { + return `CH.engine.replacingMergeTree({ version: ${str(a!)}${n === 2 ? `, isDeleted: ${str(b!)}` : ""} })` + } + if (engine.family === "CollapsingMergeTree" && n === 1) return `CH.engine.collapsingMergeTree(${str(a!)})` + if (engine.family === "VersionedCollapsingMergeTree" && n === 2) return `CH.engine.versionedCollapsingMergeTree(${str(a!)}, ${str(b!)})` + return `{ family: ${str(engine.family)}, params: [${engine.params.map(str).join(", ")}] }` +} + +const keyCode = (text: string, columns: ReadonlySet): string => { + if (text === "tuple()") return "[]" + const parts = keyParts(text) + return parts.every((part) => /^[A-Za-z_][A-Za-z0-9_]*$/.test(part) && columns.has(part)) + ? `[${parts.map(str).join(", ")}]` + : `() => [${parts.map(str).join(", ")}]` +} + +const pascal = (name: string): string => { + const words = name.split(/[^A-Za-z0-9]+/).filter(Boolean) + const joined = words.map((w) => w[0]!.toUpperCase() + w.slice(1)).join("") + return joined.length === 0 || /^\d/.test(joined) ? `T${joined}` : joined +} + +const tableCode = (out: Out, exportName: string, table: TableEntity, columns: ReadonlyArray, indexes: ReadonlyArray) => { + const names = new Set(columns.map((c) => c.name)) + const l = out.lines + l.push(`export const ${exportName} = CH.table(${str(table.name)}, {`) + l.push("\tcolumns: {") + for (const column of columns) l.push(`\t\t${key(column.name)}: ${columnCode(out, column, true)},`) + l.push("\t},") + l.push(`\tengine: ${engineCode(table.engine)},`) + if (table.orderBy !== null) l.push(`\torderBy: ${keyCode(table.orderBy, names)},`) + if (table.partitionBy !== null) l.push(`\tpartitionBy: ${str(table.partitionBy)},`) + if (table.primaryKey !== null) l.push(`\tprimaryKey: ${keyCode(table.primaryKey, names)},`) + if (table.ttl !== null) l.push(`\tttl: ${str(table.ttl)},`) + const settings = Object.entries(table.settings) + if (settings.length > 0) l.push(`\tsettings: { ${settings.map(([k, v]) => `${key(k)}: ${str(v)}`).join(", ")} },`) + if (indexes.length > 0) { + l.push("\tindexes: [") + for (const index of indexes) { + l.push(`\t\tCH.index(${str(index.name)}, ${str(index.expr)}, ${str(index.type)}, ${index.granularity}),`) + } + l.push("\t],") + } + if (table.comment !== null) l.push(`\tcomment: ${str(table.comment)},`) + l.push("})") +} + +const externalCode = (out: Out, exportName: string, table: ExternalTable) => { + out.lines.push(`/** ${table.engine}: queried, never migrated. */`) + out.lines.push(`export const ${exportName} = CH.table(${str(table.name)}, {`) + out.lines.push("\texternal: true,") + out.lines.push("\tcolumns: {") + for (const column of table.columns) out.lines.push(`\t\t${key(column.name)}: ${columnCode(out, column, false)},`) + out.lines.push("\t},") + out.lines.push("})") +} + +/** The schema module for an introspected ClickHouse database. */ +export const renderClickHouseSchema = (schema: IntrospectedSchema, options: CodegenOptions = {}): { readonly source: string; readonly todos: ReadonlyArray } => { + const out: Out = { lines: [], usesSchema: false, todos: [] } + const tables: Array = [] + const columns = new Map>() + const indexes = new Map>() + const add = (map: Map>, table: string, entity: E) => map.set(table, [...(map.get(table) ?? []), entity]) + for (const entity of schema.entities) { + if (entity.kind === "table") tables.push(entity) + else if (entity.kind === "column") add(columns, entity.table, entity) + else if (entity.kind === "index") add(indexes, entity.table, entity) + } + const taken = new Set(["CH", "Schema"]) + const exportName = (name: string) => { + const base = pascal(name) + let candidate = base + for (let i = 2; taken.has(candidate); i++) candidate = `${base}${i}` + taken.add(candidate) + return candidate + } + const objects = [ + ...tables.map((t) => ({ name: t.name, write: () => tableCode(out, exportName(t.name), t, (columns.get(t.name) ?? []).sort((a, b) => a.position - b.position), indexes.get(t.name) ?? []) })), + ...schema.external.map((t) => ({ name: t.name, write: () => externalCode(out, exportName(t.name), t) })), + ].sort((a, b) => (a.name < b.name ? -1 : a.name > b.name ? 1 : 0)) + objects.forEach((object, i) => { + if (i > 0) out.lines.push("") + object.write() + }) + + const header = [ + `// Generated by \`effect-orm pull\` from ClickHouse database \`${schema.database}\`.`, + ...(options.header ?? []).map((line) => `// ${line}`), + ...(schema.notes.length + out.todos.length > 0 ? ["//", "// Review:", ...[...schema.notes, ...out.todos].map((line) => `// - ${line}`)] : []), + ] + const imports = [ + ...(out.usesSchema ? ['import { Schema } from "effect"'] : []), + `import * as CH from ${str(options.importPath ?? "@maple-dev/effect-orm/clickhouse")}`, + ] + return { source: `${[...header, "", ...imports, "", ...out.lines].join("\n")}\n`, todos: out.todos } +} diff --git a/src/kit/generate.ts b/src/kit/generate.ts index 3fed4e7..8d88bbb 100644 --- a/src/kit/generate.ts +++ b/src/kit/generate.ts @@ -57,7 +57,8 @@ export class KitError extends Schema.TaggedError()("@maple-dev/effect- details: Schema.optional(Schema.Array(Schema.String)), }) {} -const io = (op: () => Promise, message: string) => +/** A filesystem call, failing with a `KitError` that names what it was doing. */ +export const io = (op: () => Promise, message: string) => Effect.tryPromise({ try: op, catch: (cause) => new KitError({ code: "io", message: `${message}: ${cause instanceof Error ? cause.message : String(cause)}` }) }) /** Import every schema module and collect the definitions they export. */ diff --git a/src/kit/pull.test.ts b/src/kit/pull.test.ts new file mode 100644 index 0000000..a10da01 --- /dev/null +++ b/src/kit/pull.test.ts @@ -0,0 +1,238 @@ +import { mkdtempSync, readdirSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join, resolve } from "node:path" +import { Effect, Layer } from "effect" +import { afterEach, beforeEach, describe, expect, it } from "vitest" +import { runCli } from "../kit" +import { MigrationDriver } from "../migrate/driver" +import { MigrateSqlError } from "../migrate/errors" +import { + engineSpecOf, + entitiesFromCatalog, + introspectClickHouse, + keyParts, + readClickHouseCatalog, + splitEngineFull, + stripDatabase, + type ClickHouseCatalog, +} from "../migrate/introspect" +import { entitiesOf, isSchemaObject, type SchemaObject } from "../schema/snapshot" +import { renderClickHouseSchema, typeCode } from "./codegen" + +const src = resolve(import.meta.dirname, "..") +const fixtures = resolve(import.meta.dirname, "__fixtures__") +// Read from a ClickHouse 26.2 server: the tables the live test creates, as `system.*` returned them. +const catalog = JSON.parse(readFileSync(join(fixtures, "clickhouse-catalog.json"), "utf8")) as ClickHouseCatalog + +describe("introspection", () => { + const schema = entitiesFromCatalog(catalog) + const table = (name: string) => schema.entities.find((e) => e.kind === "table" && e.name === name) + + it("reads keys as declared, not as the server extends them", () => { + expect(table("t")).toMatchObject({ + engine: { family: "ReplacingMergeTree", params: ["Ts"] }, + orderBy: "(OrgId, H, toStartOfHour(Ts), Name)", + primaryKey: "(OrgId, H)", + partitionBy: "toDate(Ts)", + ttl: "toDate(Ts) + toIntervalDay(30)", + settings: { index_granularity: "4096", storage_policy: "'default'" }, + comment: "events table", + }) + // sorting_key is `K, Ver`: the server appends the version. + expect(table("coll")).toMatchObject({ orderBy: "K", primaryKey: null, settings: {} }) + expect(table("notuple")).toMatchObject({ orderBy: "tuple()" }) + expect(table("sink")).toMatchObject({ engine: { family: "Null", params: [] }, orderBy: null }) + }) + + it("keeps views, dictionaries and the ledger out of the managed tables", () => { + const names = schema.entities.flatMap((e) => (e.kind === "table" ? [e.name] : [])) + expect(names).toEqual(["coll", "mem", "notuple", "sink", "t", "totals"]) + expect(schema.external.map((t) => [t.name, t.engine])).toEqual([ + ["dist", "Distributed"], + ["plain", "MergeTree"], + ["v", "View"], + ]) + expect(schema.notes).toEqual([ + "plain.Eph: EPHEMERAL columns are not modeled; left out", + 'plain: generated as external; table definitions take plain identifiers only, not "my-col"', + "t: SAMPLE BY H is not modeled; generated without it", + "totals_mv: materialized view not generated; write it with CH.materializedView. Its body: SELECT OrgId, Name, count() AS C FROM t GROUP BY OrgId, Name", + ]) + }) + + it("goes on without indexes when the user may not read them", async () => { + const driver = Layer.succeed(MigrationDriver, { + execute: () => Effect.void, + query: (sql: string) => + sql.includes("data_skipping_indices") + ? Effect.fail(new MigrateSqlError({ message: "Not enough privileges", sql, cause: undefined })) + : Effect.succeed( + sql.includes("currentDatabase() AS db") ? [{ db: catalog.database }] : sql.includes("system.tables") ? catalog.tables : catalog.columns, + ), + }) + const schema = await Effect.runPromise(introspectClickHouse().pipe(Effect.provide(driver))) + expect(schema.entities.some((e) => e.kind === "index")).toBe(false) + expect(schema.notes[0]).toBe("skipping indexes not read, so none are generated: Not enough privileges") + const verify = await Effect.runPromise(Effect.exit(readClickHouseCatalog().pipe(Effect.provide(driver)))) + expect(verify._tag).toBe("Failure") + }) + + it("only the tables asked for", () => { + const some = entitiesFromCatalog(catalog, { tables: ["totals", "nope"] }) + expect(some.entities.filter((e) => e.kind === "table").map((e) => e.name)).toEqual(["totals"]) + expect(some.notes).toEqual(["nope: no such table in probe"]) + }) + + it("notes what a table holds beyond the model, and writes what it cannot declare as external", () => { + const table = (name: string, engineFull: string, create = "") => ({ + name, + engine: engineFull.split("(")[0], + engine_full: engineFull, + create_table_query: create, + }) + const column = (tbl: string, name: string, kind = "") => ({ table: tbl, name, type: "String", default_kind: kind, default_expression: kind === "" ? "" : "x" }) + const schema = entitiesFromCatalog({ + database: "db", + tables: [ + table("legacy", "MergeTree(d, (k, d), 8192)"), + table("proj", "MergeTree ORDER BY k", "CREATE TABLE db.proj (k String, PROJECTION by_k (SELECT * ORDER BY k), CONSTRAINT k_set CHECK k != '') ENGINE = MergeTree ORDER BY k"), + table("eph", "MergeTree ORDER BY k"), + ], + columns: [column("legacy", "k"), column("proj", "k"), column("eph", "k"), column("eph", "raw-input", "EPHEMERAL")], + indexes: [], + }) + expect(schema.entities.filter((e) => e.kind === "table").map((e) => e.name)).toEqual(["eph", "proj"]) + expect(schema.external.map((t) => t.name)).toEqual(["legacy"]) + expect(schema.notes).toEqual([ + "legacy: generated as external; its engine is declared in the old MergeTree(date, key, granularity) syntax, which table definitions do not write", + "proj: PROJECTION by_k is not modeled; generated without it", + "proj: CONSTRAINT k_set is not modeled; generated without it", + "eph.raw-input: EPHEMERAL columns are not modeled; left out", + ]) + }) + + it("strips the database qualifier only where it is one", () => { + expect(stripDatabase("SELECT a FROM db.t JOIN `my-db`.u JOIN mydb.v JOIN x.db.w", "db")).toBe("SELECT a FROM t JOIN `my-db`.u JOIN mydb.v JOIN x.db.w") + }) + + it("splits engine_full at its clauses, not inside expressions", () => { + expect(splitEngineFull("MergeTree PARTITION BY f(' ORDER BY ') ORDER BY (a, b) TTL d + toIntervalDay(1) SETTINGS x = 1")).toEqual({ + engine: "MergeTree", + clauses: { "PARTITION BY": "f(' ORDER BY ')", "ORDER BY": "(a, b)", TTL: "d + toIntervalDay(1)", SETTINGS: "x = 1" }, + }) + expect(keyParts("(a, f(b, c))")).toEqual(["a", "f(b, c)"]) + expect(keyParts("(a + b) * c")).toEqual(["(a + b) * c"]) + }) + + it("drops the replication prefix and Keeper arguments", () => { + expect(engineSpecOf("ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/t', '{replica}', Ver)")).toEqual({ + family: "ReplacingMergeTree", + params: ["Ver"], + }) + expect(engineSpecOf("SharedMergeTree('/clickhouse/tables/{uuid}/{shard}', '{replica}')")).toEqual({ family: "MergeTree", params: [] }) + expect(engineSpecOf("ReplicatedMergeTree")).toEqual({ family: "MergeTree", params: [] }) + }) +}) + +describe("codegen", () => { + it.each([ + ["LowCardinality(Nullable(String))", "CH.lowCardinality(CH.nullable(CH.string))"], + ["Map(String, Array(Float64))", "CH.map(CH.string, CH.array(CH.float64))"], + ["Nullable(DateTime64(3))", "CH.nullable(CH.precision(CH.dateTime64, 3))"], + ["DateTime64(3, 'UTC')", 'CH.custom("DateTime64(3, \'UTC\')", Schema.String)'], + ["Enum8('it\\'s' = 1, 'b' = 2)", 'CH.custom("Enum8(\'it\\\\\'s\' = 1, \'b\' = 2)", Schema.Literals(["it\'s", "b"]))'], + ["AggregateFunction(quantiles(0.5, 0.9), UInt64)", 'CH.aggregateState("quantiles(0.5, 0.9)", "UInt64")'], + ["SimpleAggregateFunction(sum, UInt64)", 'CH.simpleAggregateFunction("sum", CH.uint64)'], + ["Decimal(9, 2)", 'CH.custom("Decimal(9, 2)", CH.CHNumber)'], + ["Map(UInt64, String)", 'CH.untyped("Map(UInt64, String)")'], + ["Tuple(a String, b UInt8)", 'CH.untyped("Tuple(a String, b UInt8)")'], + ])("%s", (sql, code) => { + expect(typeCode(sql)).toMatchObject({ code, sql }) + }) + + it("flags a decimal wider than a number holds", () => { + expect(typeCode("Decimal(9, 2)").todo).toBeUndefined() + expect(typeCode("Decimal(38, 10)").todo).toBe("Decimal(38, 10) decodes to a number, which rounds past 15 significant digits") + expect(typeCode("Decimal128(4)").todo).toBeDefined() + }) + + it("writes a module that yields back the entities it was written from", async () => { + const schema = entitiesFromCatalog(catalog) + const { source } = renderClickHouseSchema(schema, { importPath: "../../clickhouse" }) + // Committed, so `tsc` checks generated code too. + await expect(source).toMatchFileSnapshot(join(fixtures, "pulled-clickhouse.ts")) + const module = (await import(join(fixtures, "pulled-clickhouse.ts"))) as Record + const objects = Object.values(module).filter(isSchemaObject) as Array + expect(await Effect.runPromise(entitiesOf(objects))).toEqual(schema.entities) + expect(Object.keys(module).sort()).toEqual(["Coll", "Dist", "Mem", "Notuple", "Plain", "Sink", "T", "Totals", "V"]) + }) +}) + +describe("effect-orm pull", () => { + let dir = "" + let lines: Array = [] + const cli = (...args: Array) => { + const cwd = process.cwd() + process.chdir(dir) + return runCli(args, (line) => lines.push(line)).finally(() => process.chdir(cwd)) + } + + beforeEach(() => { + dir = realpathSync(mkdtempSync(join(tmpdir(), "effect-orm-pull-"))) + lines = [] + // A driver answering the catalog queries from the fixture. + writeFileSync( + join(dir, "effect-orm.config.ts"), + `import { readFileSync } from "node:fs" +import { Effect, Layer } from "effect" +import { MigrationDriver } from "${src}/migrate" +const catalog = JSON.parse(readFileSync(${JSON.stringify(join(fixtures, "clickhouse-catalog.json"))}, "utf8")) +const driver = Layer.succeed(MigrationDriver, { + execute: () => Effect.void, + query: (sql: string) => + Effect.succeed( + sql.includes("currentDatabase() AS db") ? [{ db: catalog.database }] + : sql.includes("system.tables") ? catalog.tables + : sql.includes("system.columns") ? catalog.columns + : sql.includes("system.data_skipping_indices") ? catalog.indexes + : [], + ), +}) +export default { schema: "./db/schema.ts", out: "./migrations", driver } +`, + ) + }) + afterEach(() => rmSync(dir, { recursive: true, force: true })) + + it("writes the config's schema module, which generate --baseline then adopts unchanged", async () => { + expect(await cli("pull")).toBe(0) + expect(lines[0]).toBe(`Wrote ${join(dir, "db/schema.ts")}: 6 tables, 3 external (dist, plain, v).`) + expect(lines).toContain(" t: SAMPLE BY H is not modeled; generated without it") + expect(lines).toContain(" plain.B: Map(UInt64, String) has no built-in type; rows hold it undecoded") + + // The package import, pointed at the source for this test. + const path = join(dir, "db/schema.ts") + writeFileSync(path, readFileSync(path, "utf8").replace('"@maple-dev/effect-orm/clickhouse"', JSON.stringify(`${src}/clickhouse`))) + expect(await cli("generate", "--baseline", "--name", "baseline")).toBe(0) + expect(await cli("generate")).toBe(0) + expect(lines.at(-1)).toBe("No schema changes, nothing to write.") + expect(readdirSync(join(dir, "migrations"))).toHaveLength(1) + }) + + it("takes a one-module schema array as the default --out, and refuses an empty --tables", async () => { + const config = join(dir, "effect-orm.config.ts") + writeFileSync(config, readFileSync(config, "utf8").replace('schema: "./db/schema.ts"', 'schema: ["./db/schema.ts"]')) + expect(await cli("pull", "--tables", " , ")).toBe(1) + expect(await cli("pull", "--tables", "totals")).toBe(0) + expect(readFileSync(join(dir, "db/schema.ts"), "utf8")).toContain('export const Totals = CH.table("totals", {') + }) + + it("refuses to overwrite without --force, and writes only the tables asked for", async () => { + writeFileSync(join(dir, "mine.ts"), "// mine\n") + expect(await cli("pull", "--out", "mine.ts")).toBe(1) + expect(readFileSync(join(dir, "mine.ts"), "utf8")).toBe("// mine\n") + expect(await cli("pull", "--out", "mine.ts", "--tables", "totals", "--force", "--json")).toBe(0) + expect(JSON.parse(lines.at(-1)!)).toMatchObject({ tables: ["totals"], external: [], review: [] }) + expect(readFileSync(join(dir, "mine.ts"), "utf8")).toContain('export const Totals = CH.table("totals", {') + }) +}) diff --git a/src/kit/pull.ts b/src/kit/pull.ts new file mode 100644 index 0000000..0b03f06 --- /dev/null +++ b/src/kit/pull.ts @@ -0,0 +1,58 @@ +// `effect-orm pull`: write a schema module from the database `config.driver` +// connects to, so an existing database starts with definitions that match it. +// `generate --baseline` then records them as the starting point. + +import { mkdir, stat, writeFile } from "node:fs/promises" +import { dirname, resolve } from "node:path" +import { Effect } from "effect" +import { introspectClickHouse } from "../migrate/introspect" +import { renderClickHouseSchema } from "./codegen" +import { io, KitError, type KitConfig } from "./generate" + +export interface PullOptions { + /** The file to write. */ + readonly out: string + /** Only these tables. Default: every one. */ + readonly tables?: ReadonlyArray + /** Overwrite `out` when it exists. */ + readonly force?: boolean +} + +export interface PullResult { + readonly written: string + /** Tables written with their DDL. */ + readonly tables: ReadonlyArray + /** Tables written as `external: true`: queried, never migrated. */ + readonly external: ReadonlyArray + /** What the file leaves out or could not type, one line each; also in its header. */ + readonly review: ReadonlyArray +} + +/** `effect-orm pull`. ClickHouse only, for now. */ +export const pull = (config: KitConfig, cwd: string, options: PullOptions) => + Effect.gen(function* () { + if ((config.dialect ?? "clickhouse") !== "clickhouse") { + return yield* new KitError({ code: "unsupported", message: "pull reads ClickHouse only, for now" }) + } + if (config.driver === undefined) { + return yield* new KitError({ code: "config", message: "this command needs a database: set `driver` in the config" }) + } + const path = resolve(cwd, options.out) + const exists = yield* io(() => stat(path).then(() => true, () => false), `Cannot stat ${path}`) + if (exists && options.force !== true) { + return yield* new KitError({ code: "config", message: `${options.out} exists; pass --force to overwrite it` }) + } + const schema = yield* introspectClickHouse(options.tables === undefined ? {} : { tables: options.tables }).pipe( + Effect.provide(config.driver), + Effect.mapError((error) => new KitError({ code: "io", message: `Cannot read the database: ${error instanceof Error ? error.message : String(error)}` })), + ) + const { source, todos } = renderClickHouseSchema(schema) + yield* io(() => mkdir(dirname(path), { recursive: true }), `Cannot create ${dirname(path)}`) + yield* io(() => writeFile(path, source), `Cannot write ${path}`) + return { + written: path, + tables: schema.entities.flatMap((e) => (e.kind === "table" ? [e.name] : [])), + external: schema.external.map((t) => t.name), + review: [...schema.notes, ...todos], + } satisfies PullResult + }) diff --git a/src/migrate/introspect.ts b/src/migrate/introspect.ts new file mode 100644 index 0000000..b1cebc0 --- /dev/null +++ b/src/migrate/introspect.ts @@ -0,0 +1,338 @@ +// Reading a live ClickHouse database: the catalog rows `verify` compares, and +// the same rows as entities, for `effect-orm pull`. +// +// The entities are written the way a `table` definition would produce them, +// so a definition generated from them (`kit/codegen.ts`) yields them back +// exactly: a key is split into its parts and rejoined as `renderKey` joins +// them, and an engine loses its `Replicated` / `Shared` prefix and Keeper +// arguments, which are a render option here. Keys come from `engine_full`, +// as the table was declared; expressions keep the server's own text. +// +// Only what the entity model holds is read into entities. A table with an +// engine outside the MergeTree family, `Null` and `Memory` (a view, a +// dictionary, `Distributed`, `Kafka`), or with a name that is not a plain +// identifier, comes back as an external table, with its columns only. What a +// managed table has beyond the model (a sampling key, a projection, a +// constraint, an `EPHEMERAL` column) is reported in `notes`, and so is every materialized view: its body is a +// query-DSL value, which SQL text cannot be turned into. + +import { Effect } from "effect" +import type { ColumnDefault, ColumnEntity, EngineSpec, IndexEntity, SchemaEntity, TableEntity } from "../schema/entities" +import { sortEntities } from "../schema/entities" +import { MigrationDriver } from "./driver" +import type { MigrateSqlError } from "./errors" +import { LEDGER_TABLES } from "./ledger" + +export type CatalogRow = Readonly> + +/** The rows of `system.tables`, `system.columns` and `system.data_skipping_indices` for the current database. */ +export interface ClickHouseCatalog { + readonly database: string + readonly tables: ReadonlyArray + readonly columns: ReadonlyArray + readonly indexes: ReadonlyArray + /** Why `indexes` is empty when they could not be read; see `readClickHouseCatalog`. */ + readonly indexesUnread?: string +} + +const INDEXES = `SELECT table, name, type_full, expr, granularity FROM system.data_skipping_indices WHERE database = currentDatabase() ORDER BY table, name` + +/** + * The catalog of the current database. A read-only user is often granted + * `system.tables` and `system.columns` but not `system.data_skipping_indices`; + * with `optionalIndexes` that leaves the indexes empty and says why, where + * `verify`, which would then miss drift, fails instead. + */ +export const readClickHouseCatalog = ( + options: { readonly optionalIndexes?: boolean } = {}, +): Effect.Effect => Effect.gen(function* () { + const driver = yield* MigrationDriver + const database = String((yield* driver.query("SELECT currentDatabase() AS db"))[0]?.db ?? "default") + const tables = yield* driver.query( + `SELECT name, engine, engine_full, sorting_key, partition_key, primary_key, sampling_key, as_select, create_table_query, comment FROM system.tables WHERE database = currentDatabase() AND NOT is_temporary AND NOT startsWith(name, '.inner') ORDER BY name`, + ) + const columns = yield* driver.query( + `SELECT table, name, type, default_kind, default_expression, compression_codec, comment, position FROM system.columns WHERE database = currentDatabase() ORDER BY table, position`, + ) + if (options.optionalIndexes !== true) return { database, tables, columns, indexes: yield* driver.query(INDEXES) } + return yield* driver.query(INDEXES).pipe( + Effect.map((indexes): ClickHouseCatalog => ({ database, tables, columns, indexes })), + Effect.catch((error) => Effect.succeed({ database, tables, columns, indexes: [], indexesUnread: error.message })), + ) +}) + +// SQL text, split at depth zero + +/** + * Split `text` wherever `at` matches outside parentheses, brackets and quotes. + * `at` is tried at every top-level position and returns the length it + * consumed, or 0. + */ +const splitTopLevel = (text: string, at: (rest: string) => number): Array => { + const parts: Array = [] + let depth = 0 + let quote: string | undefined + let start = 0 + for (let i = 0; i < text.length; i++) { + const char = text[i]! + if (quote !== undefined) { + if (char === "\\") i++ + else if (char === quote) quote = undefined + continue + } + if (char === "'" || char === '"' || char === "`") quote = char + else if (char === "(" || char === "[") depth++ + else if (char === ")" || char === "]") depth-- + else if (depth === 0) { + const length = at(text.slice(i)) + if (length > 0) { + parts.push(text.slice(start, i)) + start = i + length + i = start - 1 + } + } + } + parts.push(text.slice(start)) + return parts +} + +/** `a, f(b, c), d` as `["a", "f(b, c)", "d"]`; empty text is no parts. */ +export const splitList = (text: string): Array => + text.trim().length === 0 ? [] : splitTopLevel(text, (rest) => (rest[0] === "," ? 1 : 0)).map((part) => part.trim()) + +/** `Name(args)` as its name and argument list; `Name` alone has no arguments. */ +export const splitCall = (text: string): { readonly name: string; readonly args: ReadonlyArray | undefined } => { + const trimmed = text.trim() + const open = trimmed.indexOf("(") + if (open < 0 || !trimmed.endsWith(")")) return { name: trimmed, args: undefined } + return { name: trimmed.slice(0, open).trim(), args: splitList(trimmed.slice(open + 1, -1)) } +} + +const CLAUSES = ["PARTITION BY", "PRIMARY KEY", "ORDER BY", "SAMPLE BY", "TTL", "SETTINGS"] as const +type Clause = (typeof CLAUSES)[number] + +/** `engine_full` as its engine and the clauses after it. */ +export const splitEngineFull = (engineFull: string): { readonly engine: string; readonly clauses: Partial> } => { + const found: Array = [] + const parts = splitTopLevel(engineFull, (rest) => { + for (const clause of CLAUSES) { + if (rest.startsWith(` ${clause} `)) { + found.push(clause) + return clause.length + 2 + } + } + return 0 + }) + const clauses: Partial> = {} + found.forEach((clause, i) => { + clauses[clause] = parts[i + 1]!.trim() + }) + return { engine: parts[0]!.trim(), clauses } +} + +const isQuoted = (text: string): boolean => text.startsWith("'") && text.endsWith("'") + +/** An engine as the schema writes it: no `Replicated` / `Shared` prefix, no Keeper path and replica name. */ +export const engineSpecOf = (engine: string): EngineSpec => { + const call = splitCall(engine) + const name = call.name.replace(/^`(.*)`$/, "$1") + const args = call.args ?? [] + const shared = /^(Replicated|Shared)(?=\w*MergeTree$)/.exec(name) + if (shared === null) return { family: name, params: args } + const keeper = args.length >= 2 && isQuoted(args[0]!) && isQuoted(args[1]!) ? 2 : 0 + return { family: name.slice(shared[0].length), params: args.slice(keeper) } +} + +/** Whether `table` can declare the engine; the rest are external tables. */ +export const isModeledEngine = (family: string): boolean => family.endsWith("MergeTree") || family === "Null" || family === "Memory" + +/** A key as `renderKey` writes one: a single part bare, several in parentheses. */ +export const keyText = (parts: ReadonlyArray): string => (parts.length === 1 ? parts[0]! : `(${parts.join(", ")})`) + +/** A key's parts: `(a, f(b))` is two, `a` and `(a + b) * c` one. The inverse of `keyText`. */ +export const keyParts = (text: string): Array => { + const trimmed = text.trim() + if (!trimmed.startsWith("(") || !trimmed.endsWith(")")) return [trimmed] + let depth = 0 + for (let i = 0; i < trimmed.length; i++) { + if (trimmed[i] === "(") depth++ + else if (trimmed[i] === ")" && --depth === 0 && i < trimmed.length - 1) return [trimmed] + } + return splitList(trimmed.slice(1, -1)) +} + +const keyOf = (clause: string | undefined): string | null => + clause === undefined ? null : clause === "tuple()" ? "tuple()" : keyText(keyParts(clause)) + +/** `index_granularity = 8192` is the default the server writes into every MergeTree table's settings. */ +const DEFAULT_SETTINGS: Readonly> = { index_granularity: "8192" } + +const settingsOf = (text: string | undefined): Record => { + const settings: Record = {} + for (const part of splitList(text ?? "")) { + const eq = part.indexOf("=") + if (eq < 0) continue + const key = part.slice(0, eq).trim() + const value = part.slice(eq + 1).trim() + if (DEFAULT_SETTINGS[key] !== value) settings[key] = value + } + return settings +} + +const textOf = (value: unknown): string => (value === null || value === undefined ? "" : String(value)) +const orNull = (text: string): string | null => (text.length === 0 ? null : text) + +const DEFAULT_KINDS = new Set(["DEFAULT", "MATERIALIZED", "ALIAS"]) +/** The names a `table` definition accepts. */ +const PLAIN = /^[A-Za-z_][A-Za-z0-9_]*$/ + +// Entities + +/** A table `table` cannot declare, with the columns a query needs. */ +export interface ExternalTable { + readonly name: string + readonly engine: string + readonly columns: ReadonlyArray +} + +export interface IntrospectedSchema { + readonly database: string + /** Tables, their columns and their indexes, as `entitiesOf` would produce them. */ + readonly entities: ReadonlyArray + readonly external: ReadonlyArray + /** What the database holds that the entities do not, one line each. */ + readonly notes: ReadonlyArray +} + +export interface IntrospectOptions { + /** Only these tables. Default: every table but the migration ledger. */ + readonly tables?: ReadonlyArray +} + +/** `sql` without `.` qualifiers, which the server adds to a view's body. */ +export const stripDatabase = (sql: string, database: string): string => + sql.replace(new RegExp(`(?): Map> => { + const out = new Map>() + for (const row of rows) { + const table = textOf(row.table) + const list = out.get(table) + if (list === undefined) out.set(table, [row]) + else list.push(row) + } + return out +} + +/** Parts of a table's DDL the entity model does not hold, named from its `CREATE` statement. */ +const UNMODELED = [ + ["PROJECTION", /\bPROJECTION\s+`?([\w-]+)/g], + ["CONSTRAINT", /\bCONSTRAINT\s+`?([\w-]+)/g], +] as const + +/** The catalog as entities. Pure; see `introspectClickHouse`. */ +export const entitiesFromCatalog = (catalog: ClickHouseCatalog, options: IntrospectOptions = {}): IntrospectedSchema => { + const ledger = new Set(Object.values(LEDGER_TABLES)) + const wanted = options.tables === undefined ? undefined : new Set(options.tables) + const columnRows = groupByTable(catalog.columns) + const indexRows = groupByTable(catalog.indexes) + const entities: Array = [] + const external: Array = [] + const notes: Array = [] + if (catalog.indexesUnread !== undefined) { + notes.push(`skipping indexes not read, so none are generated: ${catalog.indexesUnread}`) + } + + /** A table's columns; `EPHEMERAL` ones are left out, an insert-only input no query can read. */ + const columnsOf = (table: string): Array => + (columnRows.get(table) ?? []) + .filter((row) => { + const kind = textOf(row.default_kind) + if (kind.length === 0 || DEFAULT_KINDS.has(kind)) return true + notes.push(`${table}.${textOf(row.name)}: ${kind} columns are not modeled; left out`) + return false + }) + .map((row, position): ColumnEntity => { + const kind = textOf(row.default_kind) + const codec = textOf(row.compression_codec) + return { + kind: "column", + table, + name: textOf(row.name), + position, + type: textOf(row.type), + default: kind.length === 0 ? null : { kind: kind as ColumnDefault["kind"], expr: textOf(row.default_expression) }, + codec: codec.length === 0 ? null : codec.replace(/^CODEC\((.*)\)$/s, "$1"), + comment: orNull(textOf(row.comment)), + } + }) + + for (const row of catalog.tables) { + const name = textOf(row.name) + if (ledger.has(name) || (wanted !== undefined && !wanted.has(name))) continue + const engineName = textOf(row.engine) + if (engineName === "MaterializedView") { + const body = stripDatabase(textOf(row.as_select), catalog.database) + notes.push(`${name}: materialized view not generated; write it with CH.materializedView. Its body: ${body}`) + continue + } + const { engine, clauses } = splitEngineFull(textOf(row.engine_full)) + const spec = engineSpecOf(engine.length > 0 ? engine : engineName) + const columns = columnsOf(name) + const asExternal = (reason?: string) => { + if (reason !== undefined) notes.push(`${name}: generated as external; ${reason}`) + external.push({ name, engine: engineName, columns }) + } + if (!isModeledEngine(spec.family)) { + asExternal() + continue + } + const mergeTree = spec.family.endsWith("MergeTree") + if (mergeTree && clauses["ORDER BY"] === undefined) { + asExternal("its engine is declared in the old MergeTree(date, key, granularity) syntax, which table definitions do not write") + continue + } + const unplain = [name, ...columns.map((c) => c.name)].filter((identifier) => !PLAIN.test(identifier)) + if (unplain.length > 0) { + asExternal(`table definitions take plain identifiers only, not ${unplain.map((u) => JSON.stringify(u)).join(", ")}`) + continue + } + if (textOf(row.sampling_key).length > 0) notes.push(`${name}: SAMPLE BY ${textOf(row.sampling_key)} is not modeled; generated without it`) + for (const [kind, pattern] of UNMODELED) { + for (const match of textOf(row.create_table_query).matchAll(pattern)) { + notes.push(`${name}: ${kind} ${match[1]} is not modeled; generated without it`) + } + } + const table: TableEntity = { + kind: "table", + name, + engine: spec, + // From the clauses, not `sorting_key`: a VersionedCollapsingMergeTree's has its version appended. + orderBy: mergeTree ? keyOf(clauses["ORDER BY"]) : null, + partitionBy: clauses["PARTITION BY"] ?? null, + primaryKey: mergeTree ? keyOf(clauses["PRIMARY KEY"]) : null, + ttl: clauses.TTL ?? null, + settings: settingsOf(clauses.SETTINGS), + comment: orNull(textOf(row.comment)), + } + entities.push(table, ...columns) + for (const index of indexRows.get(name) ?? []) { + entities.push({ + kind: "index", + table: name, + name: textOf(index.name), + expr: textOf(index.expr), + type: textOf(index.type_full), + granularity: Number(index.granularity), + } satisfies IndexEntity) + } + } + for (const name of wanted ?? []) { + if (!catalog.tables.some((row) => textOf(row.name) === name)) notes.push(`${name}: no such table in ${catalog.database}`) + } + return { database: catalog.database, entities: sortEntities(entities), external, notes } +} + +/** Read the current database as entities, for `effect-orm pull`. */ +export const introspectClickHouse = (options: IntrospectOptions = {}): Effect.Effect => + Effect.map(readClickHouseCatalog({ optionalIndexes: true }), (catalog) => entitiesFromCatalog(catalog, options)) diff --git a/src/migrate/verify.ts b/src/migrate/verify.ts index 23369d6..b5b02d2 100644 --- a/src/migrate/verify.ts +++ b/src/migrate/verify.ts @@ -14,6 +14,7 @@ import { Effect } from "effect" import { quoteClickHouseString } from "../sql/sql-fragment" import type { SchemaDialect, ClickHouseSnapshot, ColumnEntity, IndexEntity, MaterializedViewEntity, TableEntity } from "../schema/entities" import { MigrationDriver } from "./driver" +import { readClickHouseCatalog, splitEngineFull, stripDatabase, type CatalogRow } from "./introspect" import type { MigrateSourceError, MigrateSqlError } from "./errors" import { LEDGER_TABLES, readApplied } from "./ledger" import { readPgApplied } from "./pg-ledger" @@ -56,6 +57,19 @@ const unwrapTuple = (key: string | null): string => { const trimmed = key.trim() return trimmed.startsWith("(") && trimmed.endsWith(")") ? trimmed.slice(1, -1) : trimmed } +/** + * A table's keys as it was declared. `sorting_key` will not do: a + * VersionedCollapsingMergeTree's has its version column appended. + */ +const declaredKeys = (row: CatalogRow): Record<"sorting_key" | "partition_key" | "primary_key", string> => { + const { clauses } = splitEngineFull(String(row.engine_full ?? "")) + const sorting = clauses["ORDER BY"] === undefined ? String(row.sorting_key ?? "") : unwrapTuple(clauses["ORDER BY"]) + return { + sorting_key: sorting, + partition_key: String(row.partition_key ?? ""), + primary_key: clauses["PRIMARY KEY"] === undefined ? sorting : unwrapTuple(clauses["PRIMARY KEY"]), + } +} const engineFamily = (engine: string): string => engine.replace(/^(Replicated|Shared)(?=\w*MergeTree$)/, "") /** Server-normalized forms of SQL snippets, keyed by the input. */ @@ -115,17 +129,7 @@ export const verify = ( const verifyClickHouse = (snapshot: ClickHouseSnapshot): Effect.Effect, MigrateSqlError, MigrationDriver> => Effect.gen(function* () { const driver = yield* MigrationDriver - - const db = String((yield* driver.query("SELECT currentDatabase() AS db"))[0]?.db ?? "default") - const tables = yield* driver.query( - `SELECT name, engine, sorting_key, partition_key, primary_key, as_select, create_table_query FROM system.tables WHERE database = currentDatabase() AND NOT is_temporary AND NOT startsWith(name, '.inner')`, - ) - const columns = yield* driver.query( - `SELECT table, name, type, default_kind, default_expression FROM system.columns WHERE database = currentDatabase()`, - ) - const indexes = yield* driver.query( - `SELECT table, name, type_full, expr, granularity FROM system.data_skipping_indices WHERE database = currentDatabase()`, - ) + const { database: db, tables, columns, indexes } = yield* readClickHouseCatalog() const expectedTables = snapshot.entities.filter((e): e is TableEntity => e.kind === "table") const expectedColumns = snapshot.entities.filter((e): e is ColumnEntity => e.kind === "column") @@ -136,13 +140,13 @@ const verifyClickHouse = (snapshot: ClickHouseSnapshot): Effect.Effect [row, declaredKeys(row)])) + for (const keys of keysOf.values()) for (const key of Object.values(keys)) exprs.add(key) for (const row of columns) exprs.add(String(row.default_expression ?? "")) for (const row of indexes) exprs.add(String(row.expr ?? "")) const expr = yield* canonical(driver, "expr", exprs) - const qualified = new RegExp(`\\b${db.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}\\.`, "g") const views = new Set(expectedViews.map((v) => v.select)) - for (const row of tables) if (row.engine === "MaterializedView") views.add(String(row.as_select ?? "").replace(qualified, "")) + for (const row of tables) if (row.engine === "MaterializedView") views.add(stripDatabase(String(row.as_select ?? ""), db)) const query = yield* canonical(driver, "query", views) const types = yield* canonicalTypes(driver, new Set([...expectedColumns.map((c) => c.type), ...columns.map((c) => String(c.type))])) @@ -167,7 +171,7 @@ const verifyClickHouse = (snapshot: ClickHouseSnapshot): Effect.Effect t.name), ...expectedViews.map((v) => v.name)]) diff --git a/tests/pull.clickhouse.test.ts b/tests/pull.clickhouse.test.ts new file mode 100644 index 0000000..f68fd3e --- /dev/null +++ b/tests/pull.clickhouse.test.ts @@ -0,0 +1,92 @@ +import { mkdtempSync, rmSync } from "node:fs" +import { join } from "node:path" +import { ClickhouseClient } from "@effect/sql-clickhouse" +import { Effect, Layer } from "effect" +import { describe, expect, it } from "vitest" +import * as Kit from "@maple-dev/effect-orm/kit" +import * as Migrate from "@maple-dev/effect-orm/migrate" +import { endpoint } from "./clickhouse-support" + +const user = process.env.EFFECT_ORM_CLICKHOUSE_USER ?? "default" +const password = process.env.EFFECT_ORM_CLICKHOUSE_PASSWORD ?? "" + +const client = (database: string) => ClickhouseClient.layer({ url: endpoint!, username: user, password, database }) + +/** A database built with plain DDL, as one that predates effect-orm would be. */ +const DDL = [ + `CREATE TABLE events ( + OrgId LowCardinality(String), + Ts DateTime64(3) CODEC(Delta, ZSTD(1)), + Name String DEFAULT 'x' COMMENT 'the name', + Duration Nullable(UInt64), + Attributes Map(String, Array(Float64)), + Level Enum8('debug' = 1, 'info' = 2), + Amount Decimal(18, 4), + Id UUID, + Hour DateTime MATERIALIZED toStartOfHour(Ts), + Day Date ALIAS toDate(Ts), + INDEX idx_name Name TYPE bloom_filter(0.01) GRANULARITY 2 + ) ENGINE = ReplacingMergeTree(Ts) + PARTITION BY toDate(Ts) PRIMARY KEY OrgId ORDER BY (OrgId, toStartOfHour(Ts), Name) + TTL toDate(Ts) + INTERVAL 30 DAY SETTINGS index_granularity = 4096 COMMENT 'events table'`, + "CREATE TABLE totals (OrgId String, Name String, C UInt64) ENGINE = SummingMergeTree ORDER BY (OrgId, Name)", + "CREATE TABLE states (K String, Sign Int8, Ver UInt32, U AggregateFunction(uniq, String)) ENGINE = VersionedCollapsingMergeTree(Sign, Ver) ORDER BY K", + "CREATE TABLE scratch (A String) ENGINE = MergeTree ORDER BY tuple()", + "CREATE TABLE sink (A String) ENGINE = Null", + "CREATE MATERIALIZED VIEW totals_mv TO totals AS SELECT OrgId, Name, count() AS C FROM events GROUP BY OrgId, Name", + "CREATE VIEW recent AS SELECT OrgId, Name FROM events", +] + +describe.skipIf(!endpoint)("pull (live)", () => { + it("writes definitions that baseline cleanly and verify against the database", async () => { + const database = `eo_pull_${Date.now()}_${Math.floor(Math.random() * 1e6)}` + // Inside the package, so the generated `@maple-dev/effect-orm/clickhouse` import resolves. + const dir = mkdtempSync(join(import.meta.dirname, ".pull-")) + const driver = Layer.effect( + Migrate.MigrationDriver, + Effect.gen(function* () { + const sql = yield* ClickhouseClient.ClickhouseClient + return Migrate.fromSqlClient(sql, { command: sql.asCommand }) + }), + ).pipe(Layer.provide(client(database))) + const config: Kit.KitConfig = { schema: "./schema.ts", out: "./migrations", driver } + await Effect.runPromise( + Effect.gen(function* () { + const admin = yield* ClickhouseClient.ClickhouseClient + yield* admin.asCommand(admin.unsafe(`CREATE DATABASE ${database}`)) + yield* Effect.gen(function* () { + const sql = yield* ClickhouseClient.ClickhouseClient + for (const statement of DDL) yield* sql.asCommand(sql.unsafe(statement)) + }).pipe(Effect.provide(client(database))) + + const pulled = yield* Kit.pull(config, dir, { out: "schema.ts" }) + expect(pulled.tables).toEqual(["events", "scratch", "sink", "states", "totals"]) + expect(pulled.external).toEqual(["recent"]) + expect(pulled.review).toEqual([ + "totals_mv: materialized view not generated; write it with CH.materializedView. Its body: SELECT OrgId, Name, count() AS C FROM events GROUP BY OrgId, Name", + "events.Amount: Decimal(18, 4) decodes to a number, which rounds past 15 significant digits", + ]) + + yield* Kit.generate(config, dir, { baseline: "schema", name: "baseline" }) + expect(yield* Kit.generate(config, dir)).toEqual({ written: undefined, plan: [] }) + const migrations = yield* Kit.readMigrations(join(dir, "migrations")) + yield* Migrate.run({ migrations }).pipe(Effect.provide(driver)) + // The view and the materialized view are the database's, not the schema's. + const { drift } = yield* Migrate.verify(migrations).pipe(Effect.provide(driver)) + expect(drift).toEqual([ + { entity: "recent", problem: "unexpected" }, + { entity: "totals_mv", problem: "unexpected" }, + ]) + }).pipe( + Effect.ensuring( + Effect.gen(function* () { + const admin = yield* ClickhouseClient.ClickhouseClient + yield* Effect.orDie(admin.asCommand(admin.unsafe(`DROP DATABASE IF EXISTS ${database} SYNC`))) + }), + ), + Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true }))), + Effect.provide(client("default")), + ), + ) + }) +})