Migrate table encoding and type #11

Open
uwiger wants to merge 3 commits from uw-migration into master
4 changed files with 1639 additions and 41 deletions
+440
View File
@@ -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 refs `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** refs 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 |
File diff suppressed because it is too large Load Diff
+33 -6
View File
@@ -371,6 +371,10 @@ retry_activity(F, Alias, #{activity := #{ type := Type
return_abort(Type, error, retry_limit) return_abort(Type, error, retry_limit)
end. 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) -> retry_activity_(inner, F, Alias, Ctxt) ->
mrdb_stats:incr(Alias, inner_retries, 1), mrdb_stats:incr(Alias, inner_retries, 1),
try_f(F, Ctxt); 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); batch_if_index(Ref, insert, bag, fun insert_bag/5, Key, EncKey, EncVal, Obj, Opts);
%% insert_bag(Ref, Obj, Opts); %% insert_bag(Ref, Obj, Opts);
insert_(Ref, Key, EncKey, EncVal, Obj, Opts) -> insert_(Ref, Key, EncKey, EncVal, Obj, Opts) ->
batch_if_index(Ref, insert, set, fun insert_set/5, Key, EncKey, EncVal, Obj, Opts). %% Close over Key/Obj so dual-write can re-encode into migration target CF.
%% insert_set(Ref, Obj, Opts). 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) -> insert_set(Ref, EncKey, EncVal, _, Opts) ->
rdb_put(Ref, EncKey, EncVal, Opts). rdb_put(Ref, EncKey, EncVal, Opts).
@@ -822,6 +830,19 @@ insert_bag(Ref, EncKey, EncVal, _, Opts) ->
%% #{vsn := 1} -> %% #{vsn := 1} ->
insert_bag_v1(Ref, EncKey, EncVal, Opts). 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) -> batch_if_index(#{mode := mnesia} = Ref, _, _, F, _Key, EncKey, Data, _Obj, Opts) ->
F(Ref, EncKey, Data, undefined, Opts); F(Ref, EncKey, Data, undefined, Opts);
batch_if_index(#{name := Name, properties := #{index := [_|_] = Ixs}} = Ref, batch_if_index(#{name := Name, properties := #{index := [_|_] = Ixs}} = Ref,
@@ -1264,7 +1285,11 @@ delete(Tab, Key, Opts) ->
delete_(#{semantics := bag} = Ref, Key, EncKey, Opts) -> delete_(#{semantics := bag} = Ref, Key, EncKey, Opts) ->
batch_if_index(Ref, delete, bag, fun delete_bag/5, Key, EncKey, [], [], Opts); batch_if_index(Ref, delete, bag, fun delete_bag/5, Key, EncKey, [], [], Opts);
delete_(Ref, 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) ->
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, _) -> delete_obj_set(_, _, _, not_found, _) ->
ok; ok;
delete_obj_set(Ref, _, _, RawKey, Opts) when is_binary(RawKey) -> delete_obj_set(#{name := Name} = Ref, _, Obj, RawKey, Opts) when is_binary(RawKey) ->
rdb_delete(Ref, RawKey, Opts); rdb_delete(Ref, RawKey, Opts),
dual_delete(Ref, element(keypos(Name), Obj), Opts);
delete_obj_set(#{name := Name} = Ref, EncKey, Obj, _, Opts) -> delete_obj_set(#{name := Name} = Ref, EncKey, Obj, _, Opts) ->
case rdb_get(Ref, EncKey, []) of case rdb_get(Ref, EncKey, []) of
{ok, Bin} -> {ok, Bin} ->
Key = element(keypos(Name), Obj), Key = element(keypos(Name), Obj),
case decode_val(Bin, Key, Ref) of case decode_val(Bin, Key, Ref) of
Obj -> Obj ->
rdb_delete(Ref, EncKey, Opts); rdb_delete(Ref, EncKey, Opts),
dual_delete(Ref, Key, Opts);
_ -> _ ->
ok ok
end; end;
+231 -1
View File
@@ -16,6 +16,13 @@
manual_migration/1 manual_migration/1
, migrate_with_encoding_change/1 , migrate_with_encoding_change/1
, auto_migration/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"). -include_lib("common_test/include/ct.hrl").
@@ -31,7 +38,14 @@ all() ->
groups() -> groups() ->
[ [
{all_tests, [sequence], [ manual_migration {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) -> init_per_suite(Config) ->
@@ -125,6 +139,222 @@ migrate_with_encoding_change(_Config) ->
auto_migration(_Config) -> auto_migration(_Config) ->
ok. 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. ok({ok, Value}) -> Value.
tr_opts() -> tr_opts() ->