diff --git a/doc/online-encoding-migration.md b/doc/online-encoding-migration.md new file mode 100644 index 0000000..3bdb4fa --- /dev/null +++ b/doc/online-encoding-migration.md @@ -0,0 +1,440 @@ +# Online encoding migration (design note) + +**Status:** design +**Branch:** `uw-migration` +**Scope (MWP):** change **key/value encoding** of an existing **set** table by +copying into a new RocksDB column family while the table remains online. + +Changing Mnesia `type` (`set` ↔ `ordered_set`) is **out of scope** for the first +cut. For `mrdb` users, what matters for range seeks and key order is the +**rocksdb key encoding**, not the Mnesia metadata type. Aligning Mnesia type +via a custom schema transaction is possible later and needs separate design. + +--- + +## 1. Motivation + +Some tables were created with encodings that are awkward for production use. +Example: integer primary keys with `{term, {object, term}}` encoding. The +Erlang external term format preserves sort order for positive integers only +up to 32-bit max; beyond that keys use `SMALL_BIG_EXT` and **byte order ≠ +numeric order**. RocksDB iterators and seeks then mis-order heights. + +Historically apps worked around this by creating a second table +(`gmmp_tallies` → `gmmp_tallies2`) and migrating in application code. That is +painful and easy to get wrong. + +This feature moves **encoding migration into the backend**: dual-write to a +new column family (CF), copy old → new, then flip the live `db_ref` and drop +the old CF. + +--- + +## 2. Goals and non-goals + +### Goals (MWP) + +- Online migration of **encoding** for **set** tables (semantics `set`). +- Table remains readable and writable during migration. +- Safe under concurrent inserts/updates/deletes. +- Resume after process/node restart (without a durable RocksDB snapshot). +- Indexes (ordered) continue to work: updated on live writes as today; + index CFs are **not** dual-copied for encoding-only main-table migrate. +- Explicit start/complete as admin/schema-level operations (not silent). + +### Non-goals (MWP) + +- **Bag** tables (supported for completeness; poor fit for rocksdb dual-write). +- Changing Mnesia **type** (`set` / `ordered_set` / `bag`) in this pass. +- Rebuilding or re-encoding **index** column families. +- Multi-node coordinated migration (single backend instance first). +- Retaining a RocksDB snapshot across process restart. +- Record shape / arity / attribute renames (use `mnesia:transform_table` + or app logic). + +--- + +## 3. Existing building blocks + +| Piece | Role today | +|---|---| +| `persistent_term` meta map | `Name → db_ref()` via `put_pt` / `get_ref` / `ensure_ref` | +| CF create/drop | `create_column_family` / `drop_column_family` | +| Offline standalone→CF migrate | `migrate_standalone/2`: chunk `select`, insert new, delete old, `put_pt` | +| Snapshots | `mrdb:snapshot/1`, `release_snapshot/1` (DB-instance scoped) | +| Hot write path | `insert_` / `delete_` encode with ref’s `encoding` | + +The new feature is an **online, same-DB CF** migrate with dual-write, not a +reimplementation of standalone migration. + +--- + +## 4. Design overview + +### 4.0 Column-family naming (versions) + +RocksDB CF names must be **unique** within a DB. The existing map is: + +| Logical resource | Physical CF name | +|---|---| +| data table `Tab` | `"{d, Tab}"` (gen 0) | +| index | `"{i, Tab, I}"` | +| retainer | `"{r, Tab, R}"` | +| admin | `"{a, Alias}"` | + +That mapping has **no version**. Encoding migration needs the old and new data +CFs to coexist until cutover, so data CFs are versioned: + +| Generation | Physical CF name | +|---|---| +| 0 (legacy / first create) | `"{d, Tab}"` | +| N ≥ 1 | `"{d, Tab, N}"` | + +- `tab_to_cf_name(Tab)` still produces gen 0 for brand-new tables. +- Migration target = `data_cf_name(Tab, LiveGen + 1)`. +- Live gen is stored on the `db_ref` as `cf_gen` / `cf_name` and durably as + admin info `cf_gen` for the table. +- `cf_name_to_tab/2` maps **any** data gen back to logical `Tab`. +- On open, if several data gens exist for `Tab`, prefer the highest gen for + which we have a handle; refine with durable `cf_gen` when present (interrupted + migration may leave an incomplete higher gen — recovery should prefer the + recorded live gen once that path is fully wired). + +**Cutover does not recreate a “canonical” CF name.** The versioned target CF +*is* the new live CF; the previous generation is dropped. No second full copy. + +``` + ┌─────────────────────────────────────┐ + │ Old CF gen G (reads; encoding E_old)│ + │ name: {d,Tab} or {d,Tab,G} │ + └──────────────┬──────────────────────┘ + │ dual-write + │ (live updates) + ▼ + ┌─────────────────────────────────────┐ + │ New CF gen G+1 (encoding E_new) │ + │ name: {d,Tab,G+1} │ + │ + live dual-writes │ + │ + migrator copy (skip if exists) │ + └─────────────────────────────────────┘ + │ + ┌──────────────┴──────────────────────┐ + │ Admin info │ + │ encoding_migration meta, cf_gen │ + └─────────────────────────────────────┘ +``` + +### 4.1 Phases + +| Phase | Reads | Writes | Notes | +|---|---|---|---| +| `idle` | current CF | current CF | Normal operation | +| `dual_write` / `copying` | **old** CF only | old **and** new (re-encode) | Migrator walks old → new | +| `copy_done` | old | dual-write still on | Optional verify | +| `cutover` | **new** | **new** only | PT flip; then drop old + scratch | + +### 4.2 Ref shape during migration + +Live ref in persistent_term (conceptual): + +```erlang +OldRef#{ + migration => NewRef, %% target CF ref (encoding E_new) + migration_epoch => pos_integer(), + migration_phase => copying | copy_done +} +``` + +- **Reads / select / iterators:** use the outer (old) ref only until cutover. +- **Inserts / deletes / merges (set):** apply to old, then if `migration` + is set, re-encode and apply to `NewRef`. +- **Indexes:** `update_index` runs **once**, as today, using the live main ref + (old CF) for `read(R, Key)`. Index CF layout is independent of main + encoding (index keys are `{IxVal, Key}` under the **index** ref’s encoding). + +### 4.3 Migrator install rule + +For each key `K` observed on the migration walk: + +``` +if not present in live old(K) → skip %% deleted after snapshot / walk +else if present in new(K) → skip %% dual-write already installed latest +else put new(live_value(K)) %% re-encode with NewRef +``` + +Rationale: + +- **Skip if new has key** — dual-write wins; migrator must not clobber newer data. +- **Live re-check of old** — avoids **delete resurrection** when iterating a + snapshot that still contains `K` after a dual-delete removed it from both CFs. +- Prefer **live** value over snapshot-only value when installing. + +### 4.4 Snapshot (best-effort) + +While the migrator process lives: + +1. `Snapshot = rocksdb:snapshot(DbRef)` (same DB as old/new CFs). +2. Iterate **old** CF with `{snapshot, Snapshot}` read options from `cursor`. +3. Apply the install rule above (always re-check live old + new). + +On migrator crash/restart: snapshot is gone. Resume from scratch **cursor** +on **live** old CF with the same install rule. Correctness does not depend on +the snapshot; the snapshot only reduces iterator churn. + +### 4.5 Scratch metadata + +Either a dedicated scratch CF or admin-info keys (admin CF is enough for MWP). + +Minimum fields: + +```erlang +#{ table => tabname() + , phase => dual_write | copying | copy_done | cutover + , target => #{encoding := encoding()} %% validated + , cursor => '$first' | Key :: term() %% last successfully processed logical key + , epoch => pos_integer() + , started_at => millisecond() + } +``` + +- Cursor is the **logical** key (Erlang term), not the encoded rocksdb key. +- Resume seek on old CF uses **old** encoding: `encode_key(Cursor, OldRef)`. + +### 4.6 Cutover + +1. Ensure `phase = copy_done` (and optional verification). +2. **Barrier:** no in-flight activities that still hold a pre-migration ref + map (see §5.2), or epoch re-resolution (see below). +3. Schema/admin transaction: + - `put_pt(Tab, NewRef)` without `migration` (NewRef is gen G+1 CF) + - persist live `cf_gen = G+1` in admin info + - drop **old** generation CF only + - clear migration meta +4. Subsequent `get_ref(Tab)` returns new encoding and `cf_gen = G+1`. + +No rename and no second bulk copy: the migration target CF is permanent. + +### 4.7 Mnesia type deferred + +For `mrdb` callers, ordered seeks and range logic depend on **key encoding** +(e.g. `sext` vs `term`), not on `mnesia:table_info(Tab, type)`. + +Changing Mnesia type later may use a custom schema transaction so metadata +matches reality; that must not be conflated with CF encoding migrate and needs +extra care around `table_info`, load hooks, and any code that branches on type. + +--- + +## 5. Concurrency and correctness + +### 5.1 Dual-write before copy + +Order of start: + +1. Create empty new CF (target encoding). +2. Persist migration meta + install `migration => NewRef` on live ref (**dual-write ON**). +3. Only then start the copy walk. + +If copy ran before dual-write, concurrent writes could land only on old and be +missed or overwritten incorrectly. + +### 5.2 Stale in-process `db_ref` maps + +`ensure_ref/1` short-circuits when given a map (activities/batches hold refs). +After cutover, a process still holding the **old** map would write a CF that +is about to be dropped. + +MWP options (pick at least one): + +1. **Quiesce:** refuse cutover while table activity is non-zero / use a + short exclusive period. +2. **Epoch:** bump `migration_epoch`; `ensure_ref(Map)` re-fetches from PT if + map epoch ≠ current. +3. **Delay drop:** keep old CF read-only until epoch drained (refcounting). + +Recommended MWP: **epoch on ref + re-resolve when mismatch**, plus **delay +drop of old CF** until a quiet period or explicit admin “finalize”. + +### 5.3 Transactions and batches + +Prefer one RocksDB `WriteBatch` (same `db_ref`, two `cf_handle`s) so dual-write +insert/delete is atomic across old and new under `as_batch`. + +Index updates stay in the same batch as today (`batch_if_index`). + +### 5.4 Indexes and encoding independence + +Index CFs store `{IxVal, MainKey}` under the **index** encoding. Main encoding +change does not require rewriting index keys. Live writes continue to maintain +indexes against the logical object; after cutover, `index_read` still yields +`MainKey` terms and reads the new main CF with the new encoding. + +**Do not** dual-write index CFs in MWP encoding migration. + +### 5.5 Crash recovery + +On admin/backend restart: + +1. Reload CF handles and PT from durable admin state. +2. If migration meta says `copying` / `dual_write` / `copy_done`: + - Re-bind live ref with `migration => NewRef`. + - Resume dual-write. + - If not `copy_done`, restart migrator from `cursor` (live iterate). +3. Never drop old CF unless cutover fully committed. + +--- + +## 6. API sketch (MWP) + +```erlang +%% Start (admin / schema-level). Validates encoding; rejects bag; rejects +%% unsupported type changes. +-spec migrate_encoding(alias(), tabname(), encoding()) -> ok | {error, term()}. +migrate_encoding(Alias, Tab, NewEncoding) -> ... + +%% Optional: progress / status +-spec migration_status(tabname()) -> + idle | #{phase := atom(), cursor := term(), target := map()}. + +%% Complete cutover (if not automatic after copy_done + barrier) +-spec finalize_migration(alias(), tabname()) -> ok | {error, term()}. +``` + +Reporting can mirror `migrate_standalone/3` (`Rpt` pid + progress messages). + +--- + +## 7. Risks and open points + +| Risk | Mitigation | +|---|---| +| Delete resurrection under snapshot | Live re-check before install (§4.3) | +| Stale activity refs after cutover | Epoch re-resolve + delayed CF drop (§5.2) | +| Partial dual-write in batch | Multi-CF WriteBatch (§5.3) | +| PT lost while admin meta says migrating | Recovery rebuilds PT from admin (§5.5) | +| Huge tables | Chunked walk + cursor; progress reports | +| bag / type / index encoding | Explicitly out of MWP | + +Open (later): + +- Automatic vs manual finalize after `copy_done`. +- Optional full verification (count / sample) before cutover. +- Mnesia type alignment schema transaction. +- Bag and index-encoding migration. + +--- + +## 8. Relation to application workarounds + +Application dual tables (`*2`) remain valid for record-shape changes. Backend +encoding migration is aimed at cases like “same record, wrong key encoding” +without app-visible table rename. + +Example target: migrate `gmmp_generations` from `{term,{object,term}}` to a +sext (or other order-preserving) key encoding so height seeks stay valid past +2³² − 1 without a `generations2` table. + +--- + +# Implementation checklist + +Phased so each phase is testable. Bags and Mnesia type change stay excluded. + +## Phase 0 — Spec freeze and hooks + +- [ ] Freeze MWP scope: **set + encoding only**; document rejections for bag / type. +- [ ] Define durable migration meta schema (admin info keys vs scratch CF). +- [ ] Define public API: `migrate_encoding/3`, `migration_status/1`, + `finalize_migration/2` (names flexible). +- [ ] Define progress report messages (align with `migrate_standalone` rpt). +- [ ] List all write paths that must dual-write: insert, delete, delete_object, + update_counter/merge (if encoding-compatible), clear_table policy + (reject migrate while clearing / dual-clear both). + +## Phase 1 — Ref + dual-write plumbing + +- [ ] Extend live `db_ref` with optional `migration`, `migration_epoch`, + `migration_phase` (or keep phase only in admin meta). +- [ ] `ensure_ref/1`: when map has stale epoch, re-fetch from PT (or document + quiesce-only MWP if epoch deferred). +- [ ] Implement `dual_put` / `dual_delete` helpers: old encoding + new encoding; + same WriteBatch when `as_batch` active. +- [ ] Wire dual-write into `insert_` / `delete_` / related set paths only. +- [ ] Confirm `update_index` still runs once and reads from **old** main ref + during migration; no index dual-write. +- [ ] Unit tests: concurrent insert/delete while `migration` set updates both + CFs; index remains consistent with logical data. + +## Phase 2 — Create target CF + start migration + +- [ ] Validate `NewEncoding` via `mnesia_rocksdb_lib:check_encoding/2`. +- [ ] Reject bag semantics; reject no-op (same encoding); reject if migration + already active. +- [ ] Create empty new CF with target encoding (and consistent vsn/access_type). +- [ ] Persist migration meta; install dual-write on live ref (schema/admin txn). +- [ ] Test: after start, reads still see old data; writes appear on both CFs + (inspect via raw/ref). + +## Phase 3 — Copy walk + scratch cursor + +- [ ] Migrator process (supervised or admin-linked) with optional RocksDB snapshot. +- [ ] Iterate old CF from cursor; for each object apply install rule (§4.3): + - [ ] live old missing → skip + - [ ] new present → skip + - [ ] else put re-encoded live value to new +- [ ] Update scratch cursor after each chunk (batch cursor updates). +- [ ] Progress reporting every N keys. +- [ ] On completion set `phase = copy_done`. +- [ ] Tests: + - [ ] empty table + - [ ] populated table, no concurrent writes + - [ ] concurrent overwrites (new has key → migrator skips; final value correct) + - [ ] concurrent deletes (no resurrection) + - [ ] restart migrator mid-way (cursor resume, no snapshot) + +## Phase 4 — Recovery + +- [ ] On backend/admin start: detect incomplete migration from durable meta. +- [ ] Re-install dual-write ref; restart migrator from cursor if needed. +- [ ] Test kill -9 / admin restart mid-copy; complete and verify data. + +## Phase 5 — Cutover + drop + +- [ ] Barrier policy implemented (epoch and/or quiesce and/or delayed drop). +- [ ] `finalize_migration`: PT → NewRef only; clear migration fields. +- [ ] Drop old CF; clear scratch/meta. +- [ ] Test: post-cutover read/write/select/index_read; encoding is E_new. +- [ ] Test: in-flight activity behaviour under chosen barrier policy. + +## Phase 6 — Optional verify + polish + +- [ ] Optional key-count or sample verification under dual-write before cutover. +- [ ] Metrics: keys copied, skipped (exists), skipped (deleted), duration. +- [ ] Docs: user guide section + this design note linked from README. +- [ ] Dialyzer / property tests for encode round-trip old→new. + +## Phase 7 — Explicitly later (not MWP) + +- [ ] Mnesia `type` change via custom schema transaction. +- [ ] Bag table migration. +- [ ] Index CF encoding rebuild. +- [ ] Multi-node / multi-replica orchestration. + +--- + +## 9. Suggested first spike (days, not weeks) + +1. Dual-write only (no migrator): force `migration => NewRef` in test; prove + insert/delete hit both CFs; indexes still correct. +2. Migrator on quiescent table: full copy + cutover + drop. +3. Add concurrent writes/deletes + install rule. +4. Add cursor + restart. +5. Snapshot as optimization only. + +--- + +## 10. Document history + +| Date | Note | +|---|---| +| 2026-08-03 | Initial design from gm_mining_pool / generations encoding discussion | diff --git a/src/mnesia_rocksdb_admin.erl b/src/mnesia_rocksdb_admin.erl index dda1e16..9287f90 100644 --- a/src/mnesia_rocksdb_admin.erl +++ b/src/mnesia_rocksdb_admin.erl @@ -12,14 +12,24 @@ , related_resources/2 %% (Alias, Name) -> [RelatedTab] , prep_close/2 %% (Alias, Tab) -> ok , get_ref/1 %% (Name) -> Ref | abort() - , get_ref/2 %% (Name, Default -> Ref | Default - , request_ref/2 %% (Alias, Name) -> Ref + , get_ref/2 %% (Name, Default) -> Ref | Default + , request_ref/1 %% (Name) -> {ok, Ref} | {error, _} + , request_ref/2 %% (Alias, Name) -> {ok, Ref} | {error, _} , close_table/2 , clear_table/1 ]). -export([ migrate_standalone/2 - , migrate_standalone/3 ]). + , migrate_standalone/3 + %% Online encoding / type migration: dual-write + CF copy + , change_table_type/3 + , change_table_type/4 + , migrate_encoding/3 + , migrate_encoding/4 + , migration_status/1 + , finalize_migration/2 + , abort_migration/2 + ]). -export([ start_link/0 , init/1 @@ -49,6 +59,7 @@ backends = #{} :: #{ alias() => backend() } , standalone = #{} :: #{{alias(), table()} := cf() } , default_opts = [] :: [{atom(), _}] + , migrators = #{} :: #{ table() => pid() } }). -type st() :: #st{}. @@ -66,6 +77,18 @@ -type db_ref() :: rocksdb:db_handle(). -type properties() :: [{atom(), any()}]. +%% Options for encoding/type migration. +%% - `type': mnesia table type (set | ordered_set) +%% - `encoding': rocksdb key/value encoding +%% - `wait': when true (default for change_table_type/3), wait for copy and finalize +%% - `timeout': wait timeout (default 5 minutes) +%% - `report': print progress while waiting (default true) +-type encoding_opt() :: #{ type => set | ordered_set + , encoding => any() + , wait => boolean() + , timeout => timeout() + , report => boolean() }. + -type cf() :: mrdb:db_ref(). -type rpt() :: undefined | map(). @@ -79,6 +102,11 @@ | {write_table_property, tabname(), tuple()} | {remove_aliases, [alias()]} | {migrate, [tabname() | {tabname(), map()}], rpt()} + | {migrate_encoding, tabname(), any(), rpt()} + | {change_table_type, tabname(), encoding_opt(), rpt()} + | {migration_status, tabname()} + | {finalize_migration, tabname()} + | {abort_migration, tabname()} | {prep_close, table()} | {close_table, table()} | {clear_table, table() | cf() }. @@ -197,8 +225,32 @@ get_ref(Name) -> Other end. +%% Hot path: persistent_term. On miss (after erase_pt opens a sync window), +%% fall back to request_ref/1 so the caller waits on the admin gen_server. +%% request_ref only reads admin state — it does not put_pt; the writer that +%% erased PT is responsible for put_pt when the update completes. +%% Avoid re-entering when already inside the admin process. get_ref(Name, Default) -> - get_pt(Name, Default). + case get_pt(Name, error) of + error -> + case whereis(?MODULE) of + Pid when Pid =:= self() -> + Default; + Pid when is_pid(Pid) -> + case request_ref(Name) of + {ok, Ref} -> Ref; + _ -> Default + end; + undefined -> + Default + end; + Ref -> + Ref + end. + +%% Name-only: resolve table across aliases via admin (read-only; no put_pt). +request_ref(Name) -> + call([], {get_ref, Name}). request_ref(Alias, Name) -> call(Alias, {get_ref, Name}). @@ -289,6 +341,139 @@ migrate_standalone(Alias, Tabs, Rpt0) -> end, call(Alias, {migrate, Tabs, Rpt}). +%% @doc Change mnesia table `type' and/or rocksdb encoding. +%% +%% `Opts' may include: +%% +%% +%% If only `type' changes and the current encoding already matches the +%% default for the new type, only the mnesia schema / admin metadata are +%% updated. Otherwise an online CF encoding migration is started. +-spec change_table_type(Alias, Tab, Opts) -> ok | {error, any()} + when Alias :: alias() + , Tab :: tabname() + , Opts :: encoding_opt(). +change_table_type(Alias, Tab, Opts) when is_map(Opts) -> + Wait = maps:get(wait, Opts, true), + change_table_type_(Alias, Tab, Opts, Wait, undefined). + +%% @doc Like {@link change_table_type/3}, with an explicit progress report target. +%% Default for `wait' is `false' (start only) when `Rpt' is given. +-spec change_table_type(Alias, Tab, Opts, Rpt) -> ok | {error, any()} + when Alias :: alias() + , Tab :: tabname() + , Opts :: encoding_opt() + , Rpt :: rpt() | pid() | atom(). +change_table_type(Alias, Tab, Opts, Rpt) when is_map(Opts) -> + Wait = maps:get(wait, Opts, false), + change_table_type_(Alias, Tab, Opts, Wait, Rpt). + +change_table_type_(Alias, Tab, Opts, Wait, Rpt0) -> + MigOpts = maps:with([type, encoding], Opts), + {Rpt, WaitAlias} = prepare_mig_rpt(Rpt0, Wait, migrate_encoding), + case call(Alias, {change_table_type, Tab, MigOpts, Rpt}) of + ok when Wait -> + wait_and_maybe_finalize(Alias, Tab, Opts, WaitAlias); + Other -> + Other + end. + +%% @doc Start online encoding migration (dual-write + CF copy). Does not wait +%% for completion; call {@link finalize_migration/2} after copy_done. +%% `Encoding' may be an encoding tuple or an options map. +-spec migrate_encoding(alias(), tabname(), Encoding) -> + ok | {error, term()} + when Encoding :: encoding_opt() | any(). +migrate_encoding(Alias, Tab, Encoding) -> + migrate_encoding(Alias, Tab, Encoding, undefined). + +-spec migrate_encoding(alias(), tabname(), Encoding, rpt() | pid() | atom()) -> + ok | {error, term()} + when Encoding :: encoding_opt() | any(). +migrate_encoding(Alias, Tab, Encoding, Rpt0) -> + Opts = normalize_mig_opts(Encoding), + {Rpt, _} = prepare_mig_rpt(Rpt0, false, migrate_encoding), + call(Alias, {migrate_encoding, Tab, Opts, Rpt}). + +prepare_mig_rpt(undefined, false, _Tag) -> + {undefined, undefined}; +prepare_mig_rpt(undefined, true, Tag) -> + A = erlang:alias(), + {#{to => A, tag => Tag}, A}; +prepare_mig_rpt(To, Wait, Tag) when is_pid(To); is_atom(To) -> + prepare_mig_rpt(#{to => To, tag => Tag}, Wait, Tag); +prepare_mig_rpt(#{to := To} = R, _Wait, _Tag) -> + {R, To}; +prepare_mig_rpt(#{} = R, true, Tag) -> + A = erlang:alias(), + {R#{to => A, tag => Tag}, A}; +prepare_mig_rpt(#{} = R, false, Tag) -> + {R#{tag => maps:get(tag, R, Tag)}, undefined}. + +normalize_mig_opts(Opts) when is_map(Opts) -> + maps:with([type, encoding], Opts); +normalize_mig_opts(Encoding) -> + #{encoding => Encoding}. + +wait_and_maybe_finalize(Alias, Tab, Opts, WaitAlias) -> + case migration_status(Tab) of + idle -> + %% Schema-only type change (no CF copy) already completed. + maybe_unalias(WaitAlias), + ok; + #{} -> + Timeout = maps:get(timeout, Opts, timer:minutes(5)), + Report = maps:get(report, Opts, true), + CollectOpts0 = #{ success => encoding_migrator_done + , failure => encoding_migrator_failed + , report => Report + , timeout => Timeout }, + CollectOpts = case WaitAlias of + undefined -> CollectOpts0; + A -> CollectOpts0#{alias => A} + end, + case collect_reports(CollectOpts) of + ok -> + finalize_migration(Alias, Tab); + error -> + {error, migration_failed_or_timeout} + end + end. + +maybe_unalias(undefined) -> ok; +maybe_unalias(A) -> erlang:unalias(A). + +%% Status lives in admin CF (`encoding_migration` info), not in persistent_term. +%% Dual-write is indicated by `migration` on the live ref; phase/progress is +%% durable admin metadata updated by the migrator without touching PT. +-spec migration_status(tabname()) -> idle | map(). +migration_status(Tab) when is_atom(Tab) -> + case get_ref(Tab, error) of + #{migration := _, alias := Alias} -> + case read_info(Alias, Tab, encoding_migration, undefined) of + #{} = Meta -> Meta; + undefined -> #{phase => copying} + end; + _ -> + idle + end. + +-spec finalize_migration(alias(), tabname()) -> ok | {error, term()}. +finalize_migration(Alias, Tab) -> + call(Alias, {finalize_migration, Tab}). + +-spec abort_migration(alias(), tabname()) -> ok | {error, term()}. +abort_migration(Alias, Tab) -> + call(Alias, {abort_migration, Tab}). + -spec call(alias() | [], req()) -> no_return() | any(). call(Alias, Req) -> call(Alias, Req, infinity). @@ -423,6 +608,14 @@ handle_call({[], {add_aliases, Aliases}}, _From, St) -> handle_call({[], {remove_aliases, Aliases}}, _From, St) -> St1 = do_remove_aliases(Aliases, St), {reply, ok, St1}; +handle_call({[], {get_ref, Name}}, _From, St) -> + %% request_ref/1: read-only lookup. Writers own erase_pt/put_pt. + case find_cf(Name, St) of + #{status := open} = Ref -> + {reply, {ok, Ref}, St}; + _ -> + {reply, {error, not_found}, St} + end; handle_call({Alias, Req}, _From, St) -> handle_call_for_alias(Alias, Req, St); handle_call(_Req, _From, St) -> @@ -451,6 +644,15 @@ handle_info({mnesia_table_event, Event}, St) -> _ -> {noreply, St} end; +handle_info({'DOWN', _MRef, process, Pid, Reason}, + #st{migrators = Ms} = St) -> + case lists:keyfind(Pid, 2, maps:to_list(Ms)) of + {Tab, Pid} -> + ?log(info, "Encoding migrator for ~p exited: ~p", [Tab, Reason]), + {noreply, St#st{migrators = maps:remove(Tab, Ms)}}; + false -> + {noreply, St} + end; handle_info(_Msg, St) -> {noreply, St}. @@ -556,6 +758,8 @@ handle_req(Alias, {clear_table, Name}, Backend, #st{} = St) -> {reply, {error, not_found}, St} end; handle_req(Alias, {get_ref, Name}, Backend, #st{} = St) -> + %% Read-only: do not put_pt here. The metadata writer that called + %% erase_pt is responsible for put_pt when the update completes. case find_cf(Alias, Name, Backend, St) of {ok, #{status := open} = Ref} -> {reply, {ok, Ref}, St}; @@ -569,9 +773,10 @@ handle_req(_Alias, {related_resources, Tab}, Backend, St) -> {reply, Res, St}; handle_req(Alias, {write_table_property, Tab, Prop}, Backend, St) -> case find_cf(Alias, Tab, Backend, St) of - {ok, #{status := opens} = Cf0} -> + {ok, #{status := open} = Cf0} -> case mnesia_schema:schema_transaction( fun() -> + %% Open sync window first (schema + admin meta inside). erase_pt(Tab), Cf = update_user_properties(Prop, Cf0), St1 = update_cf(Alias, Tab, Cf, St), @@ -593,16 +798,48 @@ handle_req(Alias, {migrate, Tabs0, Rpt}, Backend, St) -> {reply, Res, St1}; {error, _} = Error -> {reply, Error, St} - end. + end; +handle_req(Alias, {change_table_type, Tab, Opts, Rpt}, Backend, St) -> + case start_change_table_type(Alias, Tab, Opts, Rpt, Backend, St) of + {ok, St1} -> + {reply, ok, St1}; + {error, _} = Error -> + {reply, Error, St} + end; + + +handle_req(Alias, {migrate_encoding, Tab, Encoding, Rpt}, Backend, St) -> + case start_encoding_migration(Alias, Tab, Encoding, Rpt, Backend, St) of + {ok, St1} -> + {reply, ok, St1}; + {error, _} = Error -> + {reply, Error, St} + end; +handle_req(Alias, {finalize_migration, Tab}, Backend, St) -> + case do_finalize_encoding_migration(Alias, Tab, Backend, St) of + {ok, St1} -> + {reply, ok, St1}; + {error, _} = Error -> + {reply, Error, St} + end; +handle_req(Alias, {abort_migration, Tab}, Backend, St) -> + {Res, St1} = do_abort_migration(Alias, Tab, Backend, St), + {reply, Res, St1}; +handle_req(_Alias, {migration_status, Tab}, _Backend, St) -> + {reply, migration_status(Tab), St}. handle_load_table_req(Alias, Name, TRec, Backend, St) -> case create_table_from_trec(Alias, Name, TRec, Backend, St) of {ok, TRec1, St1} -> TRec2 = TRec1#{status => open}, - St2 = update_cf(Alias, Name, TRec2, St1), + Backend1 = maps:get(Alias, St1#st.backends, Backend), + {TRec3, St2} = + maybe_resume_encoding_migration( + Alias, Name, TRec2, Backend1, St1), + St3 = update_cf(Alias, Name, TRec3, St2), ?log(debug, "Table loaded ~p", [Name]), - put_pt(Name, TRec2), - {reply, {ok, TRec2}, St2}; + put_pt(Name, TRec3), + {reply, {ok, TRec3}, St3}; {error, _} = Error -> {reply, Error, St} end. @@ -815,21 +1052,68 @@ prepare_migration(Alias, Tabs, Rpt, St) -> Res1 = add_related_tabs(Res, maps:get(Alias, St#st.backends), Alias, St), case [E || {error, _} = E <- Res1] of [] -> - rpt(Rpt, "Will migrate ~p~n", [[T || {T,_,_} <- Res1]]), + rpt(Rpt, will_migrate, "Will migrate ~p~n", [[T || {T,_,_} <- Res1]]), {ok, Res1}; [_|_] = Errors -> - rpt(Rpt, "Errors encountered: ~p~n", [Errors]), + rpt(Rpt, error, "Errors encountered: ~p~n", [Errors]), {error, Errors} end. rpt(Rpt, Fmt, Args) -> - rpt(Rpt, erlang:system_time(millisecond), Fmt, Args). + rpt(Rpt, undefined, Fmt, Args). -rpt(undefined, _, _, _) -> ok; -rpt(#{to := Rpt} = R, Time, Fmt, Args) -> - Rpt ! {mnesia_rocksdb, report, R#{time => Time, fmt => Fmt, args => Args}}, +rpt(Rpt, Lbl, Fmt, Args) -> + rpt(Rpt, Lbl, erlang:system_time(millisecond), Fmt, Args). + +rpt(undefined, _, _, _, _) -> ok; +rpt(#{to := Rpt} = R, Lbl, Time, Fmt, Args) -> + Rpt ! {mnesia_rocksdb, report, R#{label => Lbl, time => Time, fmt => Fmt, args => Args}}, ok. +collect_reports(#{ success := SuccessLbl + , failure := ErrorLbl } = Opts) -> + Timeout = maps:get(timeout, Opts, infinity), + receive + {mnesia_rocksdb, report, R} -> + Opts1 = maybe_print(R, Opts), + case R of + #{label := SuccessLbl} -> + collect_ret(ok, Opts1); + #{label := ErrorLbl} -> + collect_ret(error, Opts1); + R -> + collect_reports(Opts1) + end + after Timeout -> + collect_ret(error, Opts) + end. + +maybe_print(progress, #{report := true} = Opts) -> + io:fwrite(".", []), + Opts#{last => progress}; +maybe_print(#{fmt := Fmt, args := Args}, #{report := true} = Opts) -> + case Opts of + #{last := progress} -> + io:fwrite("~n" ++ nl(Fmt), Args); + _ -> + io:fwrite(nl(Fmt), Args) + end, + Opts#{last => fmt}; +maybe_print(_, Opts) -> + Opts. + +nl(S) -> + unicode:characters_to_list([string:chomp(S),"\n"]). + +collect_ret(Ret, Opts) -> + case maps:find(alias, Opts) of + {ok, A} -> + erlang:unalias(A), + Ret; + error -> + Ret + end. + maybe_progress(#{to := To}, C) when C rem 100000 =:= 0 -> To ! {mnesia_rocksdb, report, progress}; maybe_progress(_, _) -> @@ -868,14 +1152,14 @@ do_migrate_tabs(Alias, Tabs, Backend, Rpt, St) -> do_migrate_table(Alias, {Name, OldTRec, TRec0}, Backend, Rpt, St) when is_map(TRec0) -> T0 = erlang:system_time(millisecond), - rpt(Rpt, T0, "Migrate ~p~n", [Name]), + rpt(Rpt, migrating, T0, "Migrate ~p~n", [Name]), TRec = maps:without([encoding, vsn], TRec0), maybe_write_user_props(TRec), {ok, CF, St1} = create_cf_and_migrate(Alias, Name, OldTRec, TRec, Backend, Rpt, St), put_pt(Name, CF), T1 = erlang:system_time(millisecond), - rpt(Rpt, T1, "~nDone (~p)~n", [Name]), + rpt(Rpt, migration_done, T1, "~nDone (~p)~n", [Name]), Time = T1 - T0, io:fwrite("~p migrated, ~p ms~n", [Name, Time]), {{Name, {ok, Time}}, St1}. @@ -1045,6 +1329,46 @@ try_refresh_cf(#{alias := Alias, name := Name, properties := Ps} = Cf, Props, St try_refresh_cf(_, _, _) -> false. +%% Update admin cf metadata. Call erase_pt before this when opening a larger +%% sync window (e.g. around a schema transaction). Erase here too so standalone +%% use still forces concurrent get_ref callers through request_ref. +update_table_properties(Alias, Tab, Meta, St) -> + case find_cf_from_state(Alias, Tab, St) of + {ok, Cf} -> + erase_pt(Tab), + Cf1 = update_cf_props(Meta, Cf), + St1 = update_cf(Alias, Tab, Cf1, St), + put_pt(Tab, Cf1), + St1; + {error, _} -> + error({unknown_cf, {Alias, Tab}}) + end. + +update_cf_props(Meta, #{properties := Props, name := Name} = Cf) -> + Props1 = lists:foldl(fun update_cf_prop_/2, Props, Meta), + Cf1 = Cf#{properties := Props1}, + %% Keep top-level semantics aligned with mnesia type when type changes. + case {is_atom(Name), lists:keyfind(type, 1, Meta)} of + {true, {type, Type}} -> + Cf1#{semantics => Type}; + _ -> + Cf1 + end; +update_cf_props(Meta, #{properties := Props} = Cf) -> + Props1 = lists:foldl(fun update_cf_prop_/2, Props, Meta), + Cf#{properties := Props1}. + +update_cf_prop_({user_properties, UPs}, Ps) -> + %% Store full property tuples (#{Key => Prop}) like props_to_map/2 and + %% update_user_properties/2 — not bare values from a proplist merge. + UPs0 = maps:get(user_properties, Ps, #{}), + UPs1 = maps:merge( + UPs0, + maps:from_list([{element(1, P), P} || P <- UPs])), + Ps#{user_properties => UPs1}; +update_cf_prop_({K, V}, Ps) -> + Ps#{K => V}. + update_user_properties(Prop, #{properties := Ps} = Cf) -> Key = element(1, Prop), UserProps = case maps:find(user_properties, Ps) of @@ -1158,8 +1482,11 @@ create_table_as_cf(Alias, Name, #{db_ref := DbRef} = R, St) -> CfName = tab_to_cf_name(Name), case create_column_family(DbRef, CfName, cfopts(), R) of {ok, CfH} -> - R1 = check_version_and_encoding(R#{ cf_handle => CfH - , type => column_family }), + R1 = check_version_and_encoding( + R#{ cf_handle => CfH + , type => column_family + , cf_gen => maps:get(cf_gen, R, 0) + , cf_name => CfName }), {ok, R1, update_cf(Alias, Name, R1, St)}; {error, _} = Error -> Error @@ -1363,37 +1690,611 @@ admin_cfs({info, _} = I) -> [ {tab_to_cf_name(I), cfopts()} ]. map_cfs({ok, Ref, CfHandles}, CFs, Alias, Acc) -> ZippedCFs = lists:zip(CFs, CfHandles), - %% io:fwrite("ZippedCFs = ~p~n", [ZippedCFs]), - CfInfo = maps:from_list( - [{cf_name_to_tab(N, Alias), #{ db_ref => Ref - , cf_handle => H - , alias => Alias - , status => pre_existing - , type => column_family }} - || {{N,_}, H} <- ZippedCFs]), + %% Group versioned data CFs by logical tab and generation. cf_info keeps one + %% provisional live handle per logical name (highest gen); full map is in + %% data_cf_by_gen for migration resume (source gen may be lower than target). + {CfInfo, ByGen} = + lists:foldl( + fun({{N, _}, H}, {Map, BG}) -> + Logical = cf_name_to_tab(N, Alias), + Gen = case cf_name_to_data_gen(N) of + {ok, _, G} -> G; + error -> 0 + end, + Rec = #{ db_ref => Ref + , cf_handle => H + , alias => Alias + , status => pre_existing + , type => column_family + , cf_gen => Gen + , cf_name => N }, + BG1 = case is_atom(Logical) of + true -> + GMap0 = maps:get(Logical, BG, #{}), + BG#{Logical => GMap0#{Gen => Rec}}; + false -> + BG + end, + Map1 = case maps:find(Logical, Map) of + {ok, #{cf_gen := OldG}} + when is_integer(OldG), OldG > Gen -> + Map; + _ -> + Map#{Logical => Rec} + end, + {Map1, BG1} + end, {#{}, #{}}, ZippedCFs), {ok, Acc#{ db_ref => Ref - , cf_info => CfInfo }}. + , cf_info => CfInfo + , data_cf_by_gen => ByGen }}. -tab_to_cf_name(Tab) when is_atom(Tab) -> write_term({d, Tab}); +%% Column-family naming +%% --------------------- +%% RocksDB requires unique CF names within a DB. Logical table names (atoms) +%% are not enough once we rewrite a table into a new CF (encoding migration): +%% the old and new CFs must coexist until cutover. +%% +%% Data CFs are versioned: +%% gen 0 (legacy / first create): "{d, Tab}" +%% gen N (N >= 1): "{d, Tab, N}" +%% +%% The live generation is stored on the db_ref as `cf_gen` and durably in +%% admin info `{cf_gen, Tab}`. `cf_name_to_tab/2` maps any generation back to +%% the logical table name for open/recovery; when several gens exist, the +%% recorded live gen wins (see map_cfs / resolve_data_cf). +%% +%% Index / retainer / admin CFs are unchanged: one logical resource ↔ one CF. + +tab_to_cf_name(Tab) when is_atom(Tab) -> + data_cf_name(Tab, 0); tab_to_cf_name({admin, Alias}) -> write_term({a, Alias}); tab_to_cf_name({info, Tab}) -> write_term({n, Tab}); tab_to_cf_name({Tab, index, I}) -> write_term({i, Tab, I}); tab_to_cf_name({Tab, retainer, R}) -> write_term({r, Tab, R}). +%% Physical CF name for a data table generation. +data_cf_name(Tab, 0) when is_atom(Tab) -> + write_term({d, Tab}); +data_cf_name(Tab, Gen) when is_atom(Tab), is_integer(Gen), Gen > 0 -> + write_term({d, Tab, Gen}). + write_term(T) -> lists:flatten(io_lib:fwrite("~w", [T])). cf_name_to_tab(Cf, Alias) -> case read_term(Cf) of - {ok, {d, Table}} -> Table; - {ok, {i, Table, I}} -> {Table, index, I}; - {ok, {r, Table, R}} -> {Table, retainer, R}; - {ok, {n, Table}} -> {info, Table}; - {ok, {a, Alias}} -> {admin, Alias}; + {ok, {d, Table}} -> Table; + {ok, {d, Table, _Gen}} -> Table; %% versioned data CF → logical tab + {ok, {i, Table, I}} -> {Table, index, I}; + {ok, {r, Table, R}} -> {Table, retainer, R}; + {ok, {n, Table}} -> {info, Table}; + {ok, {a, Alias}} -> {admin, Alias}; _ -> {ext, Alias, Cf} end. +%% Parse generation from a data CF name string; non-data → error. +cf_name_to_data_gen(Cf) -> + case read_term(Cf) of + {ok, {d, Table}} -> {ok, Table, 0}; + {ok, {d, Table, Gen}} -> {ok, Table, Gen}; + _ -> + error + end. + +%% ===================================================================== +%% Online encoding migration (set tables) +%% ===================================================================== + +start_change_table_type(Alias, Tab, Opts0, Rpt, Backend, St) -> + Opts = normalize_mig_opts(Opts0), + case find_cf(Alias, Tab, Backend, St) of + {ok, #{status := open} = Ref} -> + start_change_table_type_(Alias, Tab, Opts, Rpt, Ref, Backend, St); + {ok, _} -> + {error, not_open}; + error -> + {error, not_found} + end. + +start_change_table_type_(Alias, Tab, Opts, Rpt, Ref, Backend, St) -> + Props = maps:get(properties, Ref), + CurType = maps:get(type, Props), + CurEnc = maps:get(encoding, Ref), + As = maps:get(attributes, Props), + NewType = maps:get(type, Opts, CurType), + if NewType =/= set, NewType =/= ordered_set -> + {error, invalid_type}; + true -> + DefaultEnc = mnesia_rocksdb_lib:default_encoding(Tab, NewType, As), + %% Type-only: migrate encoding when current differs from default + %% for the new type; otherwise schema/metadata only. + NewEnc0 = maps:get(encoding, Opts, + case NewType of + CurType -> CurEnc; + _ when CurEnc =:= DefaultEnc -> CurEnc; + _ -> DefaultEnc + end), + case mnesia_rocksdb_lib:check_encoding(NewEnc0, As) of + {ok, NewEnc} when NewEnc =/= CurEnc -> + MigOpts = Opts#{type => NewType, encoding => NewEnc}, + start_encoding_migration( + Alias, Tab, MigOpts, Rpt, Backend, St); + {ok, _NewEnc} when NewType =/= CurType -> + Meta = [{type, NewType}], + case update_mnesia_schema( + Tab, Meta, + fun() -> + {ok, update_table_properties( + Alias, Tab, Meta, St)} + end) of + {ok, _} = Ok -> Ok; + {error, _} = Err -> Err; + Other -> {error, Other} + end; + {ok, _} -> + {error, no_change}; + {error, _} = Err -> + Err + end + end. + +start_encoding_migration(Alias, Tab, Encoding0, Rpt, Backend, St) + when is_atom(Tab) -> + case find_cf(Alias, Tab, Backend, St) of + {ok, #{status := open, semantics := bag}} -> + {error, bag_not_supported}; + {ok, #{status := open, migration := _}} -> + {error, already_migrating}; + {ok, #{status := open} = OldRef} -> + case maps:get(type, OldRef, column_family) of + standalone -> + {error, standalone_not_supported}; + column_family -> + start_encoding_migration_(Alias, Tab, Encoding0, Rpt, + OldRef, Backend, St) + end; + {ok, _} -> + {error, not_open}; + error -> + {error, not_found} + end; +start_encoding_migration(_, _, _, _, _, _) -> + {error, badarg}. + +start_encoding_migration_(Alias, Tab, Opts0, Rpt, OldRef, Backend, St) -> + Opts = normalize_mig_opts(Opts0), + Props = maps:get(properties, OldRef), + As = maps:get(attributes, Props), + Enc0 = maps:get(encoding, Opts, undefined), + case mnesia_rocksdb_lib:check_encoding(Enc0, As) of + {ok, NewEnc} -> + case maps:get(encoding, OldRef) of + NewEnc -> + {error, same_encoding}; + _OldEnc -> + Schema = migration_schema_ops(NewEnc, Opts, Props), + create_and_start_encoding_mig( + Alias, Tab, NewEnc, Schema, Rpt, OldRef, Backend, St) + end; + {error, _} = Err -> + Err + end. + +migration_schema_ops(Encoding, Opts, Props) -> + S0 = [{user_properties, [{mrdb_encoding, Encoding}]}], + case Opts of + #{type := Type} when Type =/= map_get(type, Props) -> + [{type, Type}|S0]; + _ -> + S0 + end. + +create_and_start_encoding_mig(Alias, Tab, NewEnc, Schema, Rpt, OldRef, + #{db_ref := DbRef} = _Backend, St) -> + OldGen = maps:get(cf_gen, OldRef, 0), + NewGen = OldGen + 1, + CfName = data_cf_name(Tab, NewGen), + case create_column_family(DbRef, CfName, cfopts(), OldRef) of + {ok, CfH} -> + %% Open sync window before dual-write metadata is published. + %% Writers own erase_pt/put_pt; request_ref does not reinstall PT. + erase_pt(Tab), + %% Target ref: same DB, new versioned CF, new encoding; no migration field. + NewRef0 = maps:without( + [migration, migration_meta, migration_epoch], + OldRef), + NewRef0b = NewRef0#{ cf_handle => CfH + , encoding => NewEnc + , cf_gen => NewGen + , cf_name => CfName + , name => Tab + , type => column_family + , status => open }, + %% Persist encoding in user_properties *before* check_version_and_encoding + %% so it is not replaced by the table default. + NewRef1 = update_cf_props(Schema, check_version_and_encoding(NewRef0b)), + NewRef = NewRef1#{encoding => NewEnc, cf_gen => NewGen, cf_name => CfName}, + Meta = #{ phase => copying + , target => #{encoding => NewEnc, cf_gen => NewGen} + , schema => Schema + , source_gen => OldGen + , cursor => '$first' + , epoch => 1 + , started_at => erlang:system_time(millisecond) }, + %% Durable migration state (phase/progress/schema) lives in admin CF. + %% Live PT only carries dual-write target — migrator never updates PT. + write_info(Alias, Tab, encoding_migration, Meta), + %% Live ref: dual-write to NewRef; reads still use old CF handle. + %% Keep live encoding as old; schema mrdb_encoding updated on finalize. + LiveRef = OldRef#{ migration => NewRef + , migration_epoch => 1 + , cf_gen => OldGen }, + St1 = update_cf(Alias, Tab, LiveRef, St), + put_pt(Tab, LiveRef), + {Pid, _MRef} = spawn_monitor( + fun() -> + encoding_migrator(Alias, Tab, LiveRef, + NewRef, Meta, Rpt) + end), + rpt(Rpt, "Started encoding migration ~p gen ~p -> ~p (~s)~n", + [Tab, OldGen, NewGen, CfName]), + {ok, St1#st{migrators = maps:put(Tab, Pid, St#st.migrators)}}; + {error, _} = Err -> + Err + end. + +%% Re-attach dual-write (and copier) after restart when durable meta says so. +maybe_resume_encoding_migration(Alias, Name, LiveRef, Backend, St) + when is_atom(Name) -> + case read_info(Alias, Name, encoding_migration, undefined) of + #{phase := Phase, target := #{encoding := Enc, cf_gen := NewGen}} = Meta + when Phase =:= copying; Phase =:= copy_done -> + SourceGen = maps:get(source_gen, Meta, NewGen - 1), + ByGen = maps:get(data_cf_by_gen, Backend, #{}), + TabGens = maps:get(Name, ByGen, #{}), + case {maps:find(SourceGen, TabGens), maps:find(NewGen, TabGens)} of + {{ok, SrcRec}, {ok, TgtRec}} -> + resume_encoding_migration_( + Alias, Name, LiveRef, Meta, Enc, NewGen, SourceGen, + SrcRec, TgtRec, Phase, St); + _ -> + ?log(warning, + "encoding migration meta for ~p but CFs missing " + "(source=~p target=~p gens=~p)", + [Name, SourceGen, NewGen, maps:keys(TabGens)]), + {LiveRef, St} + end; + _ -> + %% Completed migration: ensure live handle matches durable cf_gen. + maybe_align_cf_gen(Alias, Name, LiveRef, Backend, St) + end; +maybe_resume_encoding_migration(_, _, LiveRef, _, St) -> + {LiveRef, St}. + +maybe_align_cf_gen(Alias, Name, LiveRef, Backend, St) -> + case read_info(Alias, Name, cf_gen, undefined) of + Gen when is_integer(Gen) -> + ByGen = maps:get(data_cf_by_gen, Backend, #{}), + case maps:find(Gen, maps:get(Name, ByGen, #{})) of + {ok, Rec} -> + Live1 = maps:merge( + LiveRef, + maps:with([cf_handle, cf_gen, cf_name, db_ref], + Rec)), + {Live1, St}; + error -> + {LiveRef, St} + end; + _ -> + {LiveRef, St} + end. + +resume_encoding_migration_(Alias, Name, LiveRef, Meta, Enc, NewGen, SourceGen, + SrcRec, TgtRec, Phase, St) -> + %% Live reads use source gen CF + old encoding (from LiveRef / schema). + Live0 = maps:merge( + LiveRef, + maps:with([cf_handle, cf_name, db_ref], SrcRec)), + Live1 = Live0#{cf_gen => SourceGen, status => open}, + NewRef0 = maps:without( + [migration, migration_meta, migration_epoch], LiveRef), + NewRef = NewRef0#{ cf_handle => maps:get(cf_handle, TgtRec) + , cf_name => maps:get(cf_name, TgtRec) + , cf_gen => NewGen + , encoding => Enc + , db_ref => maps:get(db_ref, TgtRec) + , name => Name + , status => open }, + NewRef1 = update_user_properties({mrdb_encoding, Enc}, NewRef), + NewRef2 = NewRef1#{encoding => Enc}, + Meta1 = Meta#{phase => Phase}, + %% PT: dual-write only. Phase/progress stay in admin CF (Meta1 already there). + Live2 = Live1#{ migration => NewRef2 + , migration_epoch => maps:get(epoch, Meta, 1) }, + erase_pt(Name), + St0 = update_cf(Alias, Name, Live2, St), + put_pt(Name, Live2), + St1 = case Phase of + copying -> + case maps:get(Name, St0#st.migrators, undefined) of + Pid when is_pid(Pid) -> + St0; + _ -> + {Pid, _} = spawn_monitor( + fun() -> + encoding_migrator( + Alias, Name, Live2, NewRef2, Meta1, + undefined) + end), + St0#st{migrators = maps:put(Name, Pid, St0#st.migrators)} + end; + copy_done -> + St0 + end, + {Live2, St1}. + +%% Background copier: walk old CF via mrdb:with_iterator, install into new +%% if missing (no clobber). Does not touch persistent_term. +%% +%% Durable progress is the last logical key visited (`cursor` in admin CF +%% encoding_migration info). Select continuations are not used — they are not +%% fit for persistent storage; iterator + last key is. +encoding_migrator(Alias, Tab, OldRef, NewRef, Meta0, Rpt) -> + try + Chunk = 500, + OldOnly = maps:without( + [migration, migration_meta, migration_epoch], OldRef), + Cursor0 = maps:get(cursor, Meta0, '$first'), + N0 = maps:get(copied, Meta0, 0), + N = mrdb:with_iterator( + OldOnly, + fun(I) -> + encoding_iter_loop( + I, Alias, Tab, OldOnly, NewRef, Meta0, + Cursor0, N0, Chunk, Rpt) + end), + Meta = Meta0#{ phase => copy_done + , cursor => '$end' + , copied => N + , finished_at => erlang:system_time(millisecond) }, + write_info(Alias, Tab, encoding_migration, Meta), + rpt(Rpt, encoding_migrator_done, + "Encoding migration copy done for ~p (~p objs)~n", [Tab, N]), + ok + catch + C:R:ST -> + rpt(Rpt, encoding_migrator_failed, + "encoding_migrator ~p failed: ~p:~p / ~p", + [Tab, C, R, ST]), + error({C, R}) + end. + +encoding_iter_loop(I, Alias, Tab, OldRef, NewRef, Meta0, Cursor, N, Chunk, Rpt) -> + case encoding_iter_seek(I, OldRef, Cursor) of + {ok, Obj} -> + encoding_iter_step( + I, Alias, Tab, OldRef, NewRef, Meta0, Obj, N, 0, Chunk, Rpt); + done -> + N + end. + +%% Seek to first object, or to the first object after a durable logical cursor. +encoding_iter_seek(I, _Ref, '$first') -> + case mrdb:iterator_move(I, first) of + {ok, _} = Ok -> Ok; + {error, _} -> done + end; +encoding_iter_seek(I, Ref, LastKey) -> + Enc = mnesia_rocksdb_lib:encode_key(LastKey, Ref), + case mrdb:iterator_move(I, Enc) of + {ok, Obj} -> + KP = mnesia_rocksdb_lib:keypos(maps:get(name, Ref)), + case element(KP, Obj) of + LastKey -> + case mrdb:iterator_move(I, next) of + {ok, _} = Ok -> Ok; + {error, _} -> done + end; + _ -> + %% First key >= LastKey in rocksdb order that is not LastKey + {ok, Obj} + end; + {error, _} -> + done + end. + +encoding_iter_step(I, Alias, Tab, OldRef, NewRef, Meta0, Obj, N, Since, Chunk, Rpt) -> + encoding_maybe_install(OldRef, NewRef, Obj), + N1 = N + 1, + KP = mnesia_rocksdb_lib:keypos(maps:get(name, OldRef)), + Key = element(KP, Obj), + Since1 = Since + 1, + case Since1 >= Chunk of + true -> + %% Durable logical cursor (last key successfully considered). + write_info(Alias, Tab, encoding_migration, + Meta0#{cursor => Key, copied => N1}), + maybe_progress(Rpt, N1), + encoding_iter_next( + I, Alias, Tab, OldRef, NewRef, Meta0, N1, 0, Chunk, Rpt); + false -> + encoding_iter_next( + I, Alias, Tab, OldRef, NewRef, Meta0, N1, Since1, Chunk, Rpt) + end. + +encoding_iter_next(I, Alias, Tab, OldRef, NewRef, Meta0, N, Since, Chunk, Rpt) -> + case mrdb:iterator_move(I, next) of + {ok, Obj} -> + encoding_iter_step( + I, Alias, Tab, OldRef, NewRef, Meta0, Obj, N, Since, Chunk, Rpt); + {error, _} -> + N + end. + +encoding_maybe_install(OldRef, NewRef, Obj) -> + Name = maps:get(name, OldRef), + KP = mnesia_rocksdb_lib:keypos(Name), + Key = element(KP, Obj), + OldOnly = maps:without( + [migration, migration_meta, migration_epoch], OldRef), + %% Live re-check on old (source of truth for reads during migration). + case mrdb:read(OldOnly, Key) of + [] -> + ok; %% deleted after we observed Obj + [LiveObj] -> + case mrdb:read(NewRef, Key) of + [_] -> + ok; %% dual-write already installed + [] -> + mrdb:insert(NewRef, LiveObj) + end + end. + +do_finalize_encoding_migration(Alias, Tab, Backend, St) -> + LiveRef = live_ref_for_mig(Alias, Tab, Backend, St), + case LiveRef of + #{migration := MigRef} = LR -> + %% Phase/schema from durable admin CF (migrator never updates PT). + Meta = read_info(Alias, Tab, encoding_migration, #{}), + case maps:get(phase, Meta, undefined) of + copy_done -> + update_mnesia_schema( + Tab, + maps:get(schema, Meta, []), + fun() -> + finalize_encoding_migration_( + Alias, Tab, LR, MigRef, Meta, St) + end); + Phase -> + {error, {not_ready, Phase}} + end; + error -> + {error, not_found}; + _ -> + {error, not_migrating} + end. + +do_abort_migration(Alias, Tab, Backend, St) -> + case live_ref_for_mig(Alias, Tab, Backend, St) of + #{migration := _} = LR -> + abort_encoding_migration_(Alias, Tab, LR, St); + #{} -> + {{error, not_migrating}, St}; + error -> + {{error, not_found}, St} + end. + +live_ref_for_mig(Alias, Tab, Backend, St) -> + case get_pt(Tab, error) of + error -> + case find_cf(Alias, Tab, Backend, St) of + {ok, R} -> R; + error -> error + end; + R -> + R + end. + +%% Run F inside a schema transaction that updates table cstruct fields in Meta. +%% erase_pt is done at the start of the transaction so the schema change and +%% admin-metadata side effects in F share one get_ref/request_ref sync window. +%% F is responsible for put_pt when done (or leave PT empty only on abort paths +%% that will reinstall). +update_mnesia_schema(Tab, Meta, F) -> + case mnesia_schema:schema_transaction( + fun() -> + erase_pt(Tab), + update_mnesia_schema_(Tab, Meta, F) + end) of + {atomic, Res} -> + Res; + {aborted, Error} -> + {error, Error} + end. + +update_mnesia_schema_(Tab, Meta, F) -> + TidTs = mnesia_schema:get_tid_ts_and_lock(schema, write), + ensure_writable(schema), + Cs = mnesia_schema:incr_version(mnesia_lib:val({Tab, cstruct})), + mnesia_schema:ensure_active(Cs), + List = mnesia_schema:cs2list(Cs), + List1 = lists:foldl( + fun({user_properties, UPs}, Acc) -> + {_, OldUPs} = lists:keyfind(user_properties, 1, Acc), + UPs1 = maps:to_list( + maps:iterator( + maps:merge(maps:from_list(OldUPs), + maps:from_list(UPs)), + ordered)), + lists:keyreplace( + user_properties, 1, Acc, {user_properties, UPs1}); + ({K, V}, Acc) -> + lists:keyreplace(K, 1, Acc, {K, V}) + end, List, Meta), + mnesia_schema:insert_schema_ops(TidTs, [{op, transform, ignore, List1}]), + F(). + +%% This is not exported from mnesia_schema, so copied instead. +ensure_writable(Tab) -> + case mnesia_lib:val({Tab, where_to_write}) of + [] -> + mnesia:abort({read_only, Tab}); + _ -> + ok + end. + +%% Cutover: make the versioned target CF the live one and drop the old CF. +%% No second copy — the migration target *is* the new generation. +%% Called inside update_mnesia_schema/3, which already erase_pt'd Tab. +finalize_encoding_migration_(Alias, Tab, LiveRef, MigRef, Meta, St) -> + #{db_ref := DbRef, cf_handle := OldCfH} = LiveRef, + NewEnc = maps:get(encoding, MigRef), + NewGen = maps:get(cf_gen, MigRef, maps:get(cf_gen, LiveRef, 0) + 1), + Schema = maps:get(schema, Meta, []), + LiveNew = clear_cf_migration( + MigRef#{ name => Tab + , status => open + , encoding => NewEnc + , cf_gen => NewGen }), + LiveNew1 = update_cf_props(Schema, LiveNew), + LiveNew2 = LiveNew1#{encoding => NewEnc, cf_gen => NewGen}, + St1 = update_cf(Alias, Tab, LiveNew2, St), + %% Durable live generation + clear migration marker for restart. + write_info(Alias, Tab, cf_gen, NewGen), + delete_info(Alias, Tab, encoding_migration), + _ = maybe_write_user_props(LiveNew2), + put_pt(Tab, LiveNew2), + %% Drop previous generation CF (by handle; name may be {d,Tab} or {d,Tab,G}). + ok = rocksdb:drop_column_family(DbRef, OldCfH), + try rocksdb:destroy_column_family(DbRef, OldCfH) catch error:_ -> ok end, + drop_cached_cf(maps:get(cf_name, LiveRef, + data_cf_name(Tab, maps:get(cf_gen, LiveRef, 0))), + OldCfH), + St2 = St1#st{migrators = maps:remove(Tab, St1#st.migrators)}, + {ok, St2}. + +abort_encoding_migration_(Alias, Tab, LiveRef, St) -> + case LiveRef of + #{name := Name, migration := MigRef} -> + erase_pt(Name), + #{db_ref := DbRef, cf_handle := CfH} = MigRef, + try rocksdb:drop_column_family(DbRef, CfH) catch error:_ -> ok end, + try rocksdb:destroy_column_family(DbRef, CfH) catch error:_ -> ok end, + delete_info(Alias, Tab, encoding_migration), + NewLiveRef = clear_cf_migration(LiveRef), + St1 = update_cf(Alias, Name, NewLiveRef, St), + put_pt(Tab, NewLiveRef), + St2 = St1#st{migrators = maps:remove(Tab, St1#st.migrators)}, + {ok, St2}; + _ -> + {error, no_migration} + end. + +clear_cf_migration(Cf) -> + %% migration_meta is legacy on refs; progress/phase live in admin CF only. + maps:without([migration, migration_meta, migration_epoch], Cf). + read_term(Str) -> case erl_scan:string(Str) of {ok, Tokens, _} -> diff --git a/src/mrdb.erl b/src/mrdb.erl index 8148eda..ceee89f 100644 --- a/src/mrdb.erl +++ b/src/mrdb.erl @@ -371,6 +371,10 @@ retry_activity(F, Alias, #{activity := #{ type := Type return_abort(Type, error, retry_limit) end. +%% Ctxt maps carry rocksdb opaque handles (tx | batch). Dialyzer reports +%% Wopaque_union on try_f/2 when retry Ctxt (tx handle) is unified with the +%% batch-activity Ctxt used on the first attempt via do_activity/3. +-dialyzer({no_opaque, retry_activity_/4}). retry_activity_(inner, F, Alias, Ctxt) -> mrdb_stats:incr(Alias, inner_retries, 1), try_f(F, Ctxt); @@ -811,8 +815,12 @@ insert_(#{semantics := bag} = Ref, Key, EncKey, EncVal, Obj, Opts) -> batch_if_index(Ref, insert, bag, fun insert_bag/5, Key, EncKey, EncVal, Obj, Opts); %% insert_bag(Ref, Obj, Opts); insert_(Ref, Key, EncKey, EncVal, Obj, Opts) -> - batch_if_index(Ref, insert, set, fun insert_set/5, Key, EncKey, EncVal, Obj, Opts). - %% insert_set(Ref, Obj, Opts). + %% Close over Key/Obj so dual-write can re-encode into migration target CF. + F = fun(R, EK, EV, Ix, Os) -> + insert_set(R, EK, EV, Ix, Os), + dual_put(R, Key, Obj, Os) + end, + batch_if_index(Ref, insert, set, F, Key, EncKey, EncVal, Obj, Opts). insert_set(Ref, EncKey, EncVal, _, Opts) -> rdb_put(Ref, EncKey, EncVal, Opts). @@ -822,6 +830,19 @@ insert_bag(Ref, EncKey, EncVal, _, Opts) -> %% #{vsn := 1} -> insert_bag_v1(Ref, EncKey, EncVal, Opts). +%% During online encoding migration, mirror set put/delete onto the target CF. +dual_put(#{migration := Mig0} = R, Key, Obj, Opts) -> + Mig = maps:merge(Mig0, maps:with([activity, snapshot], R)), + rdb_put(Mig, encode_key(Key, Mig), encode_val(Obj, Mig), Opts); +dual_put(_, _, _, _) -> + ok. + +dual_delete(#{migration := Mig0} = R, Key, Opts) -> + Mig = maps:merge(Mig0, maps:with([activity, snapshot], R)), + rdb_delete(Mig, encode_key(Key, Mig), Opts); +dual_delete(_, _, _) -> + ok. + batch_if_index(#{mode := mnesia} = Ref, _, _, F, _Key, EncKey, Data, _Obj, Opts) -> F(Ref, EncKey, Data, undefined, Opts); batch_if_index(#{name := Name, properties := #{index := [_|_] = Ixs}} = Ref, @@ -1264,7 +1285,11 @@ delete(Tab, Key, Opts) -> delete_(#{semantics := bag} = Ref, Key, EncKey, Opts) -> batch_if_index(Ref, delete, bag, fun delete_bag/5, Key, EncKey, [], [], Opts); delete_(Ref, Key, EncKey, Opts) -> - batch_if_index(Ref, delete, set, fun delete_set/5, Key, EncKey, [], [], Opts). + F = fun(R, EK, D, Ix, Os) -> + delete_set(R, EK, D, Ix, Os), + dual_delete(R, Key, Os) + end, + batch_if_index(Ref, delete, set, F, Key, EncKey, [], [], Opts). delete_object(Tab, Obj) -> delete_object(Tab, Obj, []). @@ -1598,15 +1623,17 @@ do_del_obj_bag_(Sz, K, Res, Obj, #{name := Name} = Ref, I, Opts) -> delete_obj_set(_, _, _, not_found, _) -> ok; -delete_obj_set(Ref, _, _, RawKey, Opts) when is_binary(RawKey) -> - rdb_delete(Ref, RawKey, Opts); +delete_obj_set(#{name := Name} = Ref, _, Obj, RawKey, Opts) when is_binary(RawKey) -> + rdb_delete(Ref, RawKey, Opts), + dual_delete(Ref, element(keypos(Name), Obj), Opts); delete_obj_set(#{name := Name} = Ref, EncKey, Obj, _, Opts) -> case rdb_get(Ref, EncKey, []) of {ok, Bin} -> Key = element(keypos(Name), Obj), case decode_val(Bin, Key, Ref) of Obj -> - rdb_delete(Ref, EncKey, Opts); + rdb_delete(Ref, EncKey, Opts), + dual_delete(Ref, Key, Opts); _ -> ok end; diff --git a/test/mnesia_rocksdb_migration_SUITE.erl b/test/mnesia_rocksdb_migration_SUITE.erl index 7e81cc2..1b12d1e 100644 --- a/test/mnesia_rocksdb_migration_SUITE.erl +++ b/test/mnesia_rocksdb_migration_SUITE.erl @@ -16,6 +16,13 @@ manual_migration/1 , migrate_with_encoding_change/1 , auto_migration/1 + , online_encoding_migration/1 + , online_encoding_migration_restart/1 + , online_encoding_migration_interrupt/1 + , change_table_type_sync/1 + , change_table_type_async/1 + , change_table_type_schema_only/1 + , change_table_type_no_change/1 ]). -include_lib("common_test/include/ct.hrl"). @@ -31,7 +38,14 @@ all() -> groups() -> [ {all_tests, [sequence], [ manual_migration - , migrate_with_encoding_change ]} + , migrate_with_encoding_change + , online_encoding_migration + , online_encoding_migration_restart + , online_encoding_migration_interrupt + , change_table_type_sync + , change_table_type_async + , change_table_type_schema_only + , change_table_type_no_change ]} ]. init_per_suite(Config) -> @@ -125,6 +139,222 @@ migrate_with_encoding_change(_Config) -> auto_migration(_Config) -> ok. +%% Online encoding migration: term keys -> sext keys while dual-writing. +online_encoding_migration(_Config) -> + ok = create_tab(enc, [{attributes, [k, v]}]), + %% Default set encoding is {term, {value, term}} for 2-attr tables. + lists:foreach( + fun(I) -> + mrdb:insert(enc, {enc, I, I * 10}) + end, lists:seq(1, 50)), + 50 = length(mrdb:select(enc, [{'_', [], ['$_']}])), + Ref0 = mrdb:get_ref(enc), + ct:log("Before migration ref: ~p", [maps:with([encoding, type, semantics], Ref0)]), + {term, _} = maps:get(encoding, Ref0), + %% Concurrent writes during migration + ok = mnesia_rocksdb_admin:migrate_encoding( + rdb, enc, {sext, {value, term}}), + %% Dual-write should be active + #{migration := _} = mrdb:get_ref(enc), + mrdb:insert(enc, {enc, 100, 1000}), + mrdb:delete(enc, 5), + %% Wait for copy_done + ok = wait_copy_done(enc, 50), + ct:log("Status after copy: ~p", [mnesia_rocksdb_admin:migration_status(enc)]), + ok = mnesia_rocksdb_admin:finalize_migration(rdb, enc), + Ref1 = mrdb:get_ref(enc), + ct:log("After finalize ref: ~p", [maps:with([encoding, type], Ref1)]), + {sext, _} = maps:get(encoding, Ref1), + false = maps:is_key(migration, Ref1), + Objs = lists:sort(mrdb:select(enc, [{'_', [], ['$_']}])), + %% 50 - 1 deleted + 1 inserted = 50 + 50 = length(Objs), + false = lists:keymember(5, 2, Objs), + true = lists:keymember(100, 2, Objs), + {enc, 100, 1000} = lists:keyfind(100, 2, Objs), + %% sext-encoded key is readable via rdb_get + {ok, _} = mrdb:rdb_get(Ref1, sext:encode(100), []), + ok. + +wait_copy_done(_Tab, 0) -> + {error, timeout}; +wait_copy_done(Tab, N) -> + case mnesia_rocksdb_admin:migration_status(Tab) of + #{phase := copy_done} -> + ok; + _ -> + timer:sleep(50), + wait_copy_done(Tab, N - 1) + end. + +%% Complete migration, restart mnesia, verify encoding + data survive. +online_encoding_migration_restart(_Config) -> + ok = create_tab(enc_r, [{attributes, [k, v]}]), + [mrdb:insert(enc_r, {enc_r, I, I * 3}) || I <- lists:seq(1, 30)], + ok = mnesia_rocksdb_admin:migrate_encoding( + rdb, enc_r, {sext, {value, term}}), + ok = wait_copy_done(enc_r, 100), + ok = mnesia_rocksdb_admin:finalize_migration(rdb, enc_r), + {sext, _} = maps:get(encoding, mrdb:get_ref(enc_r)), + Before = lists:sort(mrdb:select(enc_r, [{'_', [], ['$_']}])), + 30 = length(Before), + %% Close and reopen DB (same VM: stop/start mnesia + wait for tables). + stopped = mnesia:stop(), + ok = mnesia:start(), + ok = mnesia:wait_for_tables([enc_r], 10000), + Ref = mrdb:get_ref(enc_r), + ct:log("After restart ref: ~p", + [maps:with([encoding, cf_gen, type], Ref)]), + {sext, _} = maps:get(encoding, Ref), + false = maps:is_key(migration, Ref), + After = lists:sort(mrdb:select(enc_r, [{'_', [], ['$_']}])), + Before = After, + {ok, _} = mrdb:rdb_get(Ref, sext:encode(1), []), + ok. + +%% Stop mid-migration (after dual-write is up), reopen, finish copy + finalize. +online_encoding_migration_interrupt(_Config) -> + ok = create_tab(enc_i, [{attributes, [k, v]}]), + [mrdb:insert(enc_i, {enc_i, I, I}) || I <- lists:seq(1, 100)], + ok = mnesia_rocksdb_admin:migrate_encoding( + rdb, enc_i, {sext, {value, term}}), + %% Dual-write active; do not wait for copy_done — interrupt promptly. + #{migration := _} = mrdb:get_ref(enc_i), + mrdb:insert(enc_i, {enc_i, 200, 200}), + mrdb:delete(enc_i, 10), + stopped = mnesia:stop(), + ok = mnesia:start(), + ok = mnesia:wait_for_tables([enc_i], 10000), + %% Migration should be re-armed (dual-write) or already copy_done. + case mnesia_rocksdb_admin:migration_status(enc_i) of + #{phase := Phase} when Phase =:= copying; Phase =:= copy_done -> + ok; + Other -> + ct:fail({expected_migration_after_restart, Other}) + end, + %% Writes during resumed dual-write + mrdb:insert(enc_i, {enc_i, 201, 201}), + ok = wait_copy_done(enc_i, 200), + ok = mnesia_rocksdb_admin:finalize_migration(rdb, enc_i), + {sext, _} = maps:get(encoding, mrdb:get_ref(enc_i)), + false = maps:is_key(migration, mrdb:get_ref(enc_i)), + Objs = lists:sort(mrdb:select(enc_i, [{'_', [], ['$_']}])), + false = lists:keymember(10, 2, Objs), + true = lists:keymember(200, 2, Objs), + true = lists:keymember(201, 2, Objs), + %% 100 - 1 delete + 2 inserts = 101 + 101 = length(Objs), + ok. + +%% ===================================================================== +%% change_table_type/3 and /4 +%% ===================================================================== + +%% change_table_type/3 defaults to wait=true: copy + finalize in one call. +%% set (term keys) -> ordered_set implies default sext key encoding migration. +change_table_type_sync(_Config) -> + ok = create_tab(ctt_s, [{attributes, [k, v]}]), + set = mnesia:table_info(ctt_s, type), + [mrdb:insert(ctt_s, {ctt_s, I, I * 2}) || I <- lists:seq(1, 40)], + Ref0 = mrdb:get_ref(ctt_s), + {term, _} = maps:get(encoding, Ref0), + set = maps:get(type, maps:get(properties, Ref0)), + ok = mnesia_rocksdb_admin:change_table_type( + rdb, ctt_s, #{type => ordered_set, report => false}), + idle = mnesia_rocksdb_admin:migration_status(ctt_s), + Ref1 = mrdb:get_ref(ctt_s), + false = maps:is_key(migration, Ref1), + {sext, _} = maps:get(encoding, Ref1), + ordered_set = maps:get(type, maps:get(properties, Ref1)), + ordered_set = maps:get(semantics, Ref1), + ordered_set = mnesia:table_info(ctt_s, type), + assert_mrdb_encoding_up(ctt_s, {sext, {value, term}}), + Objs = lists:sort(mrdb:select(ctt_s, [{'_', [], ['$_']}])), + 40 = length(Objs), + {ok, _} = mrdb:rdb_get(Ref1, sext:encode(1), []), + %% Writes after change use ordered_set / sext + ok = mrdb:insert(ctt_s, {ctt_s, 100, 200}), + {ctt_s, 100, 200} = lists:keyfind(100, 2, mrdb:select(ctt_s, [{'_', [], ['$_']}])), + {ok, _} = mrdb:rdb_get(mrdb:get_ref(ctt_s), sext:encode(100), []), + ok. + +%% change_table_type/4 defaults to wait=false: start only; finalize manually. +change_table_type_async(_Config) -> + ok = create_tab(ctt_a, [{attributes, [k, v]}]), + [mrdb:insert(ctt_a, {ctt_a, I, I}) || I <- lists:seq(1, 25)], + Self = self(), + ok = mnesia_rocksdb_admin:change_table_type( + rdb, ctt_a, #{type => ordered_set, encoding => {sext, {value, term}}}, + Self), + %% Dual-write should be on immediately; type not cut over yet. + #{migration := _} = mrdb:get_ref(ctt_a), + set = mnesia:table_info(ctt_a, type), + {term, _} = maps:get(encoding, mrdb:get_ref(ctt_a)), + mrdb:insert(ctt_a, {ctt_a, 50, 50}), + mrdb:delete(ctt_a, 3), + ok = wait_copy_done(ctt_a, 100), + ok = mnesia_rocksdb_admin:finalize_migration(rdb, ctt_a), + idle = mnesia_rocksdb_admin:migration_status(ctt_a), + Ref1 = mrdb:get_ref(ctt_a), + false = maps:is_key(migration, Ref1), + {sext, _} = maps:get(encoding, Ref1), + ordered_set = mnesia:table_info(ctt_a, type), + ordered_set = maps:get(type, maps:get(properties, Ref1)), + Objs = lists:sort(mrdb:select(ctt_a, [{'_', [], ['$_']}])), + false = lists:keymember(3, 2, Objs), + true = lists:keymember(50, 2, Objs), + %% 25 - 1 delete + 1 insert = 25 + 25 = length(Objs), + {ok, _} = mrdb:rdb_get(Ref1, sext:encode(50), []), + ok. + +%% Type change only when encoding already matches the default for the new type +%% (set with sext -> ordered_set with same sext): schema/metadata, no CF copy. +change_table_type_schema_only(_Config) -> + ok = create_tab(ctt_so, + [{attributes, [k, v]}, + {user_properties, + [{mrdb_encoding, {sext, {value, term}}}]}]), + set = mnesia:table_info(ctt_so, type), + Ref0 = mrdb:get_ref(ctt_so), + {sext, _} = maps:get(encoding, Ref0), + CfGen0 = maps:get(cf_gen, Ref0, 0), + [mrdb:insert(ctt_so, {ctt_so, I, I}) || I <- lists:seq(1, 10)], + ok = mnesia_rocksdb_admin:change_table_type( + rdb, ctt_so, #{type => ordered_set, report => false}), + idle = mnesia_rocksdb_admin:migration_status(ctt_so), + Ref1 = mrdb:get_ref(ctt_so), + false = maps:is_key(migration, Ref1), + {sext, _} = maps:get(encoding, Ref1), + CfGen0 = maps:get(cf_gen, Ref1, 0), + ordered_set = mnesia:table_info(ctt_so, type), + ordered_set = maps:get(type, maps:get(properties, Ref1)), + ordered_set = maps:get(semantics, Ref1), + 10 = length(mrdb:select(ctt_so, [{'_', [], ['$_']}])), + {ok, _} = mrdb:rdb_get(Ref1, sext:encode(1), []), + ok. + +%% Idempotent / no-op errors. +change_table_type_no_change(_Config) -> + ok = create_tab(ctt_nc, [{attributes, [k, v]}]), + set = mnesia:table_info(ctt_nc, type), + {error, no_change} = + mnesia_rocksdb_admin:change_table_type( + rdb, ctt_nc, #{type => set, report => false}), + %% Same encoding as live ref + {term, ValEnc} = maps:get(encoding, mrdb:get_ref(ctt_nc)), + {error, no_change} = + mnesia_rocksdb_admin:change_table_type( + rdb, ctt_nc, #{encoding => {term, ValEnc}, report => false}), + ok. + +assert_mrdb_encoding_up(Tab, Enc) -> + UPs = mnesia:table_info(Tab, user_properties), + {mrdb_encoding, Enc} = lists:keyfind(mrdb_encoding, 1, UPs), + #{properties := #{user_properties := UPMap}} = mrdb:get_ref(Tab), + {mrdb_encoding, Enc} = maps:get(mrdb_encoding, UPMap), + ok. + ok({ok, Value}) -> Value. tr_opts() ->