QPQ-AG/mnesia_rocksdb
11
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Migrate table encoding and type #11

Open
uwiger wants to merge 3 commits from uw-migration into master
pull from: uw-migration
merge into: QPQ-AG:master
Conversation 0 Commits 3 Files Changed 4
+1056 -29
4 changed files with 1056 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 1a99fec600 - Show all commits
doc/online-encoding-migration.md
+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 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 |
src/mnesia_rocksdb_admin.erl
+465 -14
View File
@@ -19,7 +19,13 @@
]).
-export([ migrate_standalone/2
, migrate_standalone/3 ]).
, migrate_standalone/3
%% Online encoding migration (set tables): dual-write + CF copy
, migrate_encoding/3
, migrate_encoding/4
, migration_status/1
, finalize_migration/2
]).
-export([ start_link/0
, init/1
@@ -49,6 +55,7 @@
backends = #{} :: #{ alias() => backend() }
, standalone = #{} :: #{{alias(), table()} := cf() }
, default_opts = [] :: [{atom(), _}]
, migrators = #{} :: #{ table() => pid() }
}).
-type st() :: #st{}.
@@ -79,6 +86,9 @@
| {write_table_property, tabname(), tuple()}
| {remove_aliases, [alias()]}
| {migrate, [tabname() | {tabname(), map()}], rpt()}
| {migrate_encoding, tabname(), any(), rpt()}
| {migration_status, tabname()}
| {finalize_migration, tabname()}
| {prep_close, table()}
| {close_table, table()}
| {clear_table, table() | cf() }.
@@ -289,6 +299,39 @@ migrate_standalone(Alias, Tabs, Rpt0) ->
end,
call(Alias, {migrate, Tabs, Rpt}).
%% @doc Start online encoding migration for a set table (dual-write + CF copy).
-spec migrate_encoding(alias(), tabname(), Encoding :: any()) ->
ok | {error, term()}.
migrate_encoding(Alias, Tab, Encoding) ->
migrate_encoding(Alias, Tab, Encoding, undefined).
-spec migrate_encoding(alias(), tabname(), Encoding :: any(), rpt()) ->
ok | {error, term()}.
migrate_encoding(Alias, Tab, Encoding, Rpt0) ->
Rpt = case Rpt0 of
undefined -> undefined;
To when is_pid(To); is_atom(To) ->
#{to => To, tag => migrate_encoding};
#{} = M -> M
end,
call(Alias, {migrate_encoding, Tab, Encoding, Rpt}).
-spec migration_status(tabname()) -> idle | map().
migration_status(Tab) when is_atom(Tab) ->
case get_ref(Tab, error) of
#{migration := _} = Ref ->
case maps:get(migration_meta, Ref, undefined) of
#{} = Meta -> Meta;
undefined -> #{phase => dual_write}
end;
_ ->
idle
end.
-spec finalize_migration(alias(), tabname()) -> ok | {error, term()}.
finalize_migration(Alias, Tab) ->
call(Alias, {finalize_migration, Tab}).
-spec call(alias() | [], req()) -> no_return() | any().
call(Alias, Req) ->
call(Alias, Req, infinity).
@@ -451,6 +494,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}.
@@ -593,16 +645,36 @@ handle_req(Alias, {migrate, Tabs0, Rpt}, Backend, St) ->
{reply, Res, St1};
{error, _} = Error ->
{reply, Error, St}
end.
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, {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.
@@ -1158,8 +1230,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,29 +1438,81 @@ 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
%% 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 }}
|| {{N,_}, H} <- ZippedCFs]),
, 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, {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};
@@ -1394,6 +1521,330 @@ cf_name_to_tab(Cf, 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_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, Encoding0, Rpt, OldRef, Backend, St) ->
As = maps:get(attributes,
maps:get(properties, OldRef, #{}),
[key, val]),
case mnesia_rocksdb_lib:check_encoding(Encoding0, As) of
{ok, NewEnc} ->
case maps:get(encoding, OldRef) of
NewEnc ->
{error, same_encoding};
_OldEnc ->
create_and_start_encoding_mig(
Alias, Tab, NewEnc, Rpt, OldRef, Backend, St)
end;
{error, _} = Err ->
Err
end.
create_and_start_encoding_mig(Alias, Tab, NewEnc, 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} ->
%% 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_user_properties(
{mrdb_encoding, NewEnc},
check_version_and_encoding(NewRef0b)),
NewRef = NewRef1#{encoding => NewEnc, cf_gen => NewGen, cf_name => CfName},
Meta = #{ phase => copying
, target => #{encoding => NewEnc, cf_gen => NewGen}
, source_gen => OldGen
, cursor => '$first'
, epoch => 1
, started_at => erlang:system_time(millisecond) },
%% Live ref: dual-write to NewRef; reads still use Old CF handle.
LiveRef = OldRef#{ migration => NewRef
, migration_meta => Meta
, migration_epoch => 1
, cf_gen => OldGen },
write_info(Alias, Tab, encoding_migration, Meta),
%% Keep live encoding as old for correct reads; only target has NewEnc.
%% Schema mrdb_encoding is updated on finalize.
LiveRef2 = LiveRef#{migration := NewRef},
St1 = update_cf(Alias, Tab, LiveRef2, St),
put_pt(Tab, LiveRef2),
{Pid, _MRef} = spawn_monitor(
fun() ->
encoding_migrator(Alias, Tab, LiveRef2,
NewRef, 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},
Live2 = Live1#{ migration => NewRef2
, migration_meta => Meta1
, migration_epoch => maps:get(epoch, Meta, 1) },
put_pt(Name, Live2),
St1 = case Phase of
copying ->
case maps:get(Name, St#st.migrators, undefined) of
Pid when is_pid(Pid) ->
St;
_ ->
{Pid, _} = spawn_monitor(
fun() ->
encoding_migrator(
Alias, Name, Live2, NewRef2,
undefined)
end),
St#st{migrators = maps:put(Name, Pid, St#st.migrators)}
end;
copy_done ->
St
end,
{Live2, St1}.
%% Background copier: walk old CF, install into new if missing (no clobber).
encoding_migrator(Alias, Tab, OldRef, NewRef, Rpt) ->
try
Chunk = 500,
N = encoding_copy_loop(OldRef, NewRef, '$first', 0, Chunk, Rpt),
Meta = #{ phase => copy_done
, target => #{encoding => maps:get(encoding, NewRef)}
, cursor => '$end'
, epoch => 1
, copied => N
, finished_at => erlang:system_time(millisecond) },
%% Update live PT meta to copy_done (keep dual-write).
case get_ref(Tab, error) of
#{migration := _} = Live ->
put_pt(Tab, Live#{migration_meta => Meta});
_ ->
ok
end,
write_info(Alias, Tab, encoding_migration, Meta),
rpt(Rpt, "Encoding migration copy done for ~p (~p objs)~n", [Tab, N]),
ok
catch
C:R:ST ->
?log(error, "encoding_migrator ~p failed: ~p:~p / ~p",
[Tab, C, R, ST]),
error({C, R})
end.
encoding_copy_loop(OldRef, NewRef, Cursor, N, Chunk, Rpt) ->
{Batch, Cont} = encoding_select_chunk(OldRef, Cursor, Chunk),
N1 = lists:foldl(
fun(Obj, Acc) ->
encoding_maybe_install(OldRef, NewRef, Obj),
Acc + 1
end, N, Batch),
case Cont of
'$end_of_table' ->
N1;
NextCursor ->
maybe_progress(Rpt, N1),
encoding_copy_loop(OldRef, NewRef, NextCursor, N1, Chunk, Rpt)
end.
%% Chunked select on the old CF. Continuations use mrdb:select/1 when available.
encoding_select_chunk(OldRef, '$first', Limit) ->
case mrdb:select(OldRef, [{'_', [], ['$_']}], Limit) of
{L, Cont} when is_list(L) ->
{L, {cont, Cont}};
L when is_list(L) ->
{L, '$end_of_table'};
'$end_of_table' ->
{[], '$end_of_table'}
end;
encoding_select_chunk(_OldRef, {cont, Cont}, _Limit) ->
case mrdb:select(Cont) of
{L, Cont1} when is_list(L) ->
{L, {cont, Cont1}};
L when is_list(L) ->
{L, '$end_of_table'};
'$end_of_table' ->
{[], '$end_of_table'}
end;
encoding_select_chunk(OldRef, _Other, Limit) ->
%% Fallback: restart from first (safe, may re-process; install is idempotent).
encoding_select_chunk(OldRef, '$first', Limit).
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) ->
%% Prefer PT (migrator updates phase there) over gen_server state.
case get_ref(Tab, error) of
#{migration := MigRef} = LiveRef ->
Meta = maps:get(migration_meta, LiveRef, #{}),
case maps:get(phase, Meta, undefined) of
copy_done ->
finalize_encoding_migration_(Alias, Tab, LiveRef, MigRef, St);
Phase ->
{error, {not_ready, Phase}}
end;
error ->
{error, not_found};
_ ->
{error, not_migrating}
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.
finalize_encoding_migration_(Alias, Tab, LiveRef, MigRef, 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),
LiveNew = maps:without(
[migration, migration_meta, migration_epoch],
MigRef#{ name => Tab
, status => open
, encoding => NewEnc
, cf_gen => NewGen }),
LiveNew1 = update_user_properties({mrdb_encoding, NewEnc}, LiveNew),
put_pt(Tab, LiveNew1),
St1 = update_cf(Alias, Tab, LiveNew1, St),
%% Durable live generation + schema encoding for restart.
write_info(Alias, Tab, cf_gen, NewGen),
delete_info(Alias, Tab, encoding_migration),
_ = maybe_write_user_props(LiveNew1),
%% Drop previous generation CF (by handle; name may be {d,Tab} or {d,Tab,G}).
ok = rocksdb:drop_column_family(DbRef, OldCfH),
catch rocksdb:destroy_column_family(DbRef, OldCfH),
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}.
read_term(Str) ->
case erl_scan:string(Str) of
{ok, Tokens, _} ->
src/mrdb.erl
+29 -6
View File
@@ -811,8 +811,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) ->
rdb_put(R, EK, EV, 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 +826,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 +1281,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) ->
rdb_delete(R, EK, 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 +1619,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;
test/mnesia_rocksdb_migration_SUITE.erl
+114 -1
View File
@@ -16,6 +16,9 @@
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
]).
-include_lib("common_test/include/ct.hrl").
@@ -31,7 +34,10 @@ 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 ]}
].
init_per_suite(Config) ->
@@ -125,6 +131,113 @@ 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.
ok({ok, Value}) -> Value.
tr_opts() ->