diff --git a/src/mnesia_rocksdb_admin.erl b/src/mnesia_rocksdb_admin.erl index fb14178..9287f90 100644 --- a/src/mnesia_rocksdb_admin.erl +++ b/src/mnesia_rocksdb_admin.erl @@ -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: +%% +%% +%% 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) +start_change_table_type(Alias, Tab, Opts0, Rpt, Backend, St) -> + Opts = normalize_mig_opts(Opts0), + case find_cf(Alias, Tab, Backend, St) of + {ok, #{status := open} = Ref} -> + start_change_table_type_(Alias, Tab, Opts, Rpt, Ref, Backend, St); + {ok, _} -> + {error, not_open}; + error -> + {error, not_found} + end. + +start_change_table_type_(Alias, Tab, Opts, Rpt, Ref, Backend, St) -> + Props = maps:get(properties, Ref), + CurType = maps:get(type, Props), + CurEnc = maps:get(encoding, Ref), + As = maps:get(attributes, Props), + NewType = maps:get(type, Opts, CurType), + if NewType =/= set, NewType =/= ordered_set -> + {error, invalid_type}; + true -> + DefaultEnc = mnesia_rocksdb_lib:default_encoding(Tab, NewType, As), + %% Type-only: migrate encoding when current differs from default + %% for the new type; otherwise schema/metadata only. + NewEnc0 = maps:get(encoding, Opts, + case NewType of + CurType -> CurEnc; + _ when CurEnc =:= DefaultEnc -> CurEnc; + _ -> DefaultEnc + end), + case mnesia_rocksdb_lib:check_encoding(NewEnc0, As) of + {ok, NewEnc} when NewEnc =/= CurEnc -> + MigOpts = Opts#{type => NewType, encoding => NewEnc}, + start_encoding_migration( + Alias, Tab, MigOpts, Rpt, Backend, St); + {ok, _NewEnc} when NewType =/= CurType -> + Meta = [{type, NewType}], + case update_mnesia_schema( + Tab, Meta, + fun() -> + {ok, update_table_properties( + Alias, Tab, Meta, St)} + end) of + {ok, _} = Ok -> Ok; + {error, _} = Err -> Err; + Other -> {error, Other} end; - true -> - {error, invalid_type} - 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'} - 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_iter_seek(I, Ref, LastKey) -> + Enc = mnesia_rocksdb_lib:encode_key(LastKey, Ref), + case mrdb:iterator_move(I, Enc) of + {ok, Obj} -> + KP = mnesia_rocksdb_lib:keypos(maps:get(name, Ref)), + case element(KP, Obj) of + LastKey -> + case mrdb:iterator_move(I, next) of + {ok, _} = Ok -> Ok; + {error, _} -> done + end; + _ -> + %% First key >= LastKey in rocksdb order that is not LastKey + {ok, Obj} + end; + {error, _} -> + done + end. + +encoding_iter_step(I, Alias, Tab, OldRef, NewRef, Meta0, Obj, N, Since, Chunk, Rpt) -> + encoding_maybe_install(OldRef, NewRef, Obj), + N1 = N + 1, + KP = mnesia_rocksdb_lib:keypos(maps:get(name, OldRef)), + Key = element(KP, Obj), + Since1 = Since + 1, + case Since1 >= Chunk of + true -> + %% Durable logical cursor (last key successfully considered). + write_info(Alias, Tab, encoding_migration, + Meta0#{cursor => Key, copied => N1}), + maybe_progress(Rpt, N1), + encoding_iter_next( + I, Alias, Tab, OldRef, NewRef, Meta0, N1, 0, Chunk, Rpt); + false -> + encoding_iter_next( + I, Alias, Tab, OldRef, NewRef, Meta0, N1, Since1, Chunk, Rpt) + end. + +encoding_iter_next(I, Alias, Tab, OldRef, NewRef, Meta0, N, Since, Chunk, Rpt) -> + case mrdb:iterator_move(I, next) of + {ok, Obj} -> + encoding_iter_step( + I, Alias, Tab, OldRef, NewRef, Meta0, Obj, N, Since, Chunk, Rpt); + {error, _} -> + N + end. encoding_maybe_install(OldRef, NewRef, Obj) -> Name = maps:get(name, OldRef), @@ -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) -> diff --git a/src/mrdb.erl b/src/mrdb.erl index fe5736b..ceee89f 100644 --- a/src/mrdb.erl +++ b/src/mrdb.erl @@ -371,6 +371,10 @@ retry_activity(F, Alias, #{activity := #{ type := Type return_abort(Type, error, retry_limit) end. +%% Ctxt maps carry rocksdb opaque handles (tx | batch). Dialyzer reports +%% Wopaque_union on try_f/2 when retry Ctxt (tx handle) is unified with the +%% batch-activity Ctxt used on the first attempt via do_activity/3. +-dialyzer({no_opaque, retry_activity_/4}). retry_activity_(inner, F, Alias, Ctxt) -> mrdb_stats:incr(Alias, inner_retries, 1), try_f(F, Ctxt); @@ -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). diff --git a/test/mnesia_rocksdb_migration_SUITE.erl b/test/mnesia_rocksdb_migration_SUITE.erl index c5cd3b2..1b12d1e 100644 --- a/test/mnesia_rocksdb_migration_SUITE.erl +++ b/test/mnesia_rocksdb_migration_SUITE.erl @@ -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() ->