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

Improve consistency, add test cases for change_table_type()

Browse Source
This commit is contained in:
Ulf Wiger
2026-08-10 19:37:29 +02:00
parent 9f00d75654
commit ca242b72ac
3 changed files with 495 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mnesia_rocksdb_admin.erl
+367 -145
View File
@@ -12,20 +12,23 @@
, 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
%% Online encoding migration (set tables): dual-write + CF copy
%% 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
@@ -74,8 +77,17 @@
-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() }.
, encoding => any()
, wait => boolean()
, timeout => timeout()
, report => boolean() }.
-type cf() :: mrdb:db_ref().
@@ -91,8 +103,10 @@
| {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() }.
@@ -211,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}).
@@ -303,59 +341,126 @@ migrate_standalone(Alias, Tabs, Rpt0) ->
end,
call(Alias, {migrate, Tabs, Rpt}).
%% @doc Change mnesia table `type' and/or rocksdb encoding.
%%
%% `Opts' may include:
%% <ul>
%% <li>`type' - `set' | `ordered_set'</li>
%% <li>`encoding' - target key/value encoding</li>
%% <li>`wait' - `true' (default): wait for copy and finalize;
%% `false': start dual-write + copier only; call
%% {@link finalize_migration/2} later</li>
%% <li>`timeout' - wait timeout (default 5 minutes)</li>
%% <li>`report' - print progress while waiting (default `true')</li>
%% </ul>
%%
%% 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) ->
A = erlang:alias(),
case change_table_type(Alias, Tab, Opts, #{to => A}) of
ok ->
case collect_reports(#{ success => encoding_migrator_done
, failure => encoding_migrator_failed
, alias => A
, report => true
, timeout => timer:minutes(5) }) of
ok ->
finalize_migration(Alias, Tab)
end;
{error,_} = Error ->
abort_migration(Alias, Tab),
Error
end.
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().
change_table_type(Alias, Tab, Opts, Rpt) ->
call(Alias, {change_table_type, Tab, Opts, Rpt}).
, 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).
%% @doc Start online encoding migration for a set table (dual-write + CF copy).
-spec migrate_encoding(alias(), tabname(), Encoding :: encoding_opt()) ->
ok | {error, term()}.
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 :: encoding_opt(), rpt()) ->
ok | {error, term()}.
-spec migrate_encoding(alias(), tabname(), Encoding, rpt() | pid() | atom()) ->
ok | {error, term()}
when Encoding :: encoding_opt() | any().
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}).
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 := _} = Ref ->
case maps:get(migration_meta, Ref, undefined) of
#{migration := _, alias := Alias} ->
case read_info(Alias, Tab, encoding_migration, undefined) of
#{} = Meta -> Meta;
undefined -> #{phase => dual_write}
undefined -> #{phase => copying}
end;
_ ->
idle
@@ -503,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) ->
@@ -645,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};
@@ -658,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),
@@ -1213,24 +1329,42 @@ 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),
update_cf(Alias, Tab, Cf1, St);
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(UPs)),
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}.
@@ -1652,28 +1786,57 @@ cf_name_to_data_gen(Cf) ->
%% Online encoding migration (set tables)
%% =====================================================================
start_change_table_type(Alias, Tab, Opts, Rpt, Backend, St) ->
Ref = get_ref(Tab),
case Opts of
#{type := T} when not is_map_key(encoding, Opts) ->
if T == set; T == ordered_set ->
case maps:get(type, maps:get(properties, Ref)) of
T ->
{error, same_type};
_ ->
Meta = [{type, T}],
update_mnesia_schema(
Tab, Meta,
fun() -> update_table_properties(
Alias, Tab, Meta, St)
end)
end;
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 ->
{error, invalid_type}
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;
_ ->
%% Start a migration; metadata changed when finalized
start_encoding_migration(Alias, Tab, Opts, Rpt, Backend, St)
{ok, _} ->
{error, no_change};
{error, _} = Err ->
Err
end
end.
start_encoding_migration(Alias, Tab, Encoding0, Rpt, Backend, St)
@@ -1699,7 +1862,8 @@ start_encoding_migration(Alias, Tab, Encoding0, Rpt, Backend, St)
start_encoding_migration(_, _, _, _, _, _) ->
{error, badarg}.
start_encoding_migration_(Alias, Tab, Opts, Rpt, OldRef, Backend, St) ->
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),
@@ -1733,6 +1897,9 @@ create_and_start_encoding_mig(Alias, Tab, NewEnc, Schema, Rpt, OldRef,
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],
@@ -1747,9 +1914,6 @@ create_and_start_encoding_mig(Alias, Tab, NewEnc, Schema, Rpt, OldRef,
%% 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)),
%% 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}
@@ -1758,21 +1922,19 @@ create_and_start_encoding_mig(Alias, Tab, NewEnc, Schema, Rpt, OldRef,
, cursor => '$first'
, epoch => 1
, started_at => erlang:system_time(millisecond) },
io:fwrite("Meta = ~p~n", [Meta]),
%% Live ref: dual-write to NewRef; reads still use Old CF handle.
%% 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_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),
St1 = update_cf(Alias, Tab, LiveRef, St),
put_pt(Tab, LiveRef),
{Pid, _MRef} = spawn_monitor(
fun() ->
encoding_migrator(Alias, Tab, LiveRef2,
encoding_migrator(Alias, Tab, LiveRef,
NewRef, Meta, Rpt)
end),
rpt(Rpt, "Started encoding migration ~p gen ~p -> ~p (~s)~n",
@@ -1847,15 +2009,17 @@ resume_encoding_migration_(Alias, Name, LiveRef, Meta, Enc, NewGen, SourceGen,
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_meta => Meta1
, 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, St#st.migrators, undefined) of
case maps:get(Name, St0#st.migrators, undefined) of
Pid when is_pid(Pid) ->
St;
St0;
_ ->
{Pid, _} = spawn_monitor(
fun() ->
@@ -1863,77 +2027,110 @@ resume_encoding_migration_(Alias, Name, LiveRef, Meta, Enc, NewGen, SourceGen,
Alias, Name, Live2, NewRef2, Meta1,
undefined)
end),
St#st{migrators = maps:put(Name, Pid, St#st.migrators)}
St0#st{migrators = maps:put(Name, Pid, St0#st.migrators)}
end;
copy_done ->
St
St0
end,
{Live2, St1}.
%% Background copier: walk old CF, install into new if missing (no clobber).
%% 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,
N = encoding_copy_loop(OldRef, NewRef, '$first', 0, Chunk, Rpt),
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'
, 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_migrator_done, "Encoding migration copy done for ~p (~p objs)~n", [Tab, N]),
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",
rpt(Rpt, encoding_migrator_failed,
"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)
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.
%% 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'}
%% 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_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'}
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;
encoding_select_chunk(OldRef, _Other, Limit) ->
%% Fallback: restart from first (safe, may re-process; install is idempotent).
encoding_select_chunk(OldRef, '$first', Limit).
_ ->
%% 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),
@@ -1954,18 +2151,20 @@ encoding_maybe_install(OldRef, NewRef, Obj) ->
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, #{}),
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),
maps:get(schema, Meta, []),
fun() ->
finalize_encoding_migration_(Alias, Tab, LiveRef, MigRef, St)
finalize_encoding_migration_(
Alias, Tab, LR, MigRef, Meta, St)
end);
Phase ->
{error, {not_ready, Phase}}
@@ -1976,25 +2175,42 @@ do_finalize_encoding_migration(Alias, Tab, _Backend, St) ->
{error, not_migrating}
end.
do_abort_migration(Alias, Tab, _Backend, St) ->
case get_ref(Tab, error) of
#{migration := _} = LiveRef ->
abort_encoding_migration_(Alias, Tab, LiveRef, St);
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, Error}
end.
update_mnesia_schema_(Tab, Meta, F) ->
@@ -2030,22 +2246,25 @@ ensure_writable(Tab) ->
%% 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) ->
%% 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_user_properties({mrdb_encoding, NewEnc}, LiveNew),
put_pt(Tab, LiveNew1),
St1 = update_cf(Alias, Tab, LiveNew1, St),
%% Durable live generation + schema encoding for restart.
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(LiveNew1),
_ = 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,
@@ -2061,16 +2280,19 @@ abort_encoding_migration_(Alias, Tab, LiveRef, St) ->
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),
{ok, St1};
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) ->
src/mrdb.erl
+8 -4
View File
@@ -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);
@@ -812,8 +816,8 @@ insert_(#{semantics := bag} = Ref, Key, EncKey, EncVal, Obj, Opts) ->
%% insert_bag(Ref, Obj, Opts);
insert_(Ref, Key, EncKey, EncVal, 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),
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).
@@ -1281,8 +1285,8 @@ 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) ->
F = fun(R, EK, _D, _Ix, Os) ->
rdb_delete(R, EK, Os),
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).
test/mnesia_rocksdb_migration_SUITE.erl
+118 -1
View File
@@ -19,6 +19,10 @@
, 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").
@@ -37,7 +41,11 @@ groups() ->
, migrate_with_encoding_change
, online_encoding_migration
, online_encoding_migration_restart
, online_encoding_migration_interrupt ]}
, 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) ->
@@ -238,6 +246,115 @@ online_encoding_migration_interrupt(_Config) ->
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() ->