From 9f00d756544a2bfb6eeef01c51dd4f758355d9ac Mon Sep 17 00:00:00 2001 From: Ulf Wiger Date: Mon, 10 Aug 2026 17:26:47 +0200 Subject: [PATCH] WIP implement change_table_type/3 --- src/mnesia_rocksdb_admin.erl | 300 ++++++++++++++++++++++++++++++----- 1 file changed, 264 insertions(+), 36 deletions(-) diff --git a/src/mnesia_rocksdb_admin.erl b/src/mnesia_rocksdb_admin.erl index 32ea43f..fb14178 100644 --- a/src/mnesia_rocksdb_admin.erl +++ b/src/mnesia_rocksdb_admin.erl @@ -21,6 +21,7 @@ -export([ migrate_standalone/2 , migrate_standalone/3 %% Online encoding migration (set tables): dual-write + CF copy + , change_table_type/3 , migrate_encoding/3 , migrate_encoding/4 , migration_status/1 @@ -73,6 +74,9 @@ -type db_ref() :: rocksdb:db_handle(). -type properties() :: [{atom(), any()}]. +-type encoding_opt() :: #{ type => set | ordered_set + , encoding => any() }. + -type cf() :: mrdb:db_ref(). -type rpt() :: undefined | map(). @@ -299,13 +303,42 @@ migrate_standalone(Alias, Tabs, Rpt0) -> end, call(Alias, {migrate, Tabs, Rpt}). +-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. + +-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}). + %% @doc Start online encoding migration for a set table (dual-write + CF copy). --spec migrate_encoding(alias(), tabname(), Encoding :: any()) -> +-spec migrate_encoding(alias(), tabname(), Encoding :: encoding_opt()) -> ok | {error, term()}. migrate_encoding(Alias, Tab, Encoding) -> migrate_encoding(Alias, Tab, Encoding, undefined). --spec migrate_encoding(alias(), tabname(), Encoding :: any(), rpt()) -> +-spec migrate_encoding(alias(), tabname(), Encoding :: encoding_opt(), rpt()) -> ok | {error, term()}. migrate_encoding(Alias, Tab, Encoding, Rpt0) -> Rpt = case Rpt0 of @@ -332,6 +365,10 @@ migration_status(Tab) when is_atom(Tab) -> finalize_migration(Alias, Tab) -> call(Alias, {finalize_migration, Tab}). +-spec abort_migration(alias(), tabname()) -> ok | {error, term()}. +abort_migration(Alias, Tab) -> + call(Alias, {abort_migration, Tab}). + -spec call(alias() | [], req()) -> no_return() | any(). call(Alias, Req) -> call(Alias, Req, infinity). @@ -646,6 +683,15 @@ handle_req(Alias, {migrate, Tabs0, Rpt}, Backend, St) -> {error, _} = Error -> {reply, Error, St} end; +handle_req(Alias, {change_table_type, Tab, Opts, Rpt}, Backend, St) -> + case start_change_table_type(Alias, Tab, Opts, Rpt, Backend, St) of + {ok, St1} -> + {reply, ok, St1}; + {error, _} = Error -> + {reply, Error, St} + end; + + handle_req(Alias, {migrate_encoding, Tab, Encoding, Rpt}, Backend, St) -> case start_encoding_migration(Alias, Tab, Encoding, Rpt, Backend, St) of {ok, St1} -> @@ -660,6 +706,9 @@ handle_req(Alias, {finalize_migration, Tab}, Backend, St) -> {error, _} = Error -> {reply, Error, St} end; +handle_req(Alias, {abort_migration, Tab}, Backend, St) -> + {Res, St1} = do_abort_migration(Alias, Tab, Backend, St), + {reply, Res, St1}; handle_req(_Alias, {migration_status, Tab}, _Backend, St) -> {reply, migration_status(Tab), St}. @@ -887,21 +936,68 @@ prepare_migration(Alias, Tabs, Rpt, St) -> Res1 = add_related_tabs(Res, maps:get(Alias, St#st.backends), Alias, St), case [E || {error, _} = E <- Res1] of [] -> - rpt(Rpt, "Will migrate ~p~n", [[T || {T,_,_} <- Res1]]), + rpt(Rpt, will_migrate, "Will migrate ~p~n", [[T || {T,_,_} <- Res1]]), {ok, Res1}; [_|_] = Errors -> - rpt(Rpt, "Errors encountered: ~p~n", [Errors]), + rpt(Rpt, error, "Errors encountered: ~p~n", [Errors]), {error, Errors} end. rpt(Rpt, Fmt, Args) -> - rpt(Rpt, erlang:system_time(millisecond), Fmt, Args). + rpt(Rpt, undefined, Fmt, Args). -rpt(undefined, _, _, _) -> ok; -rpt(#{to := Rpt} = R, Time, Fmt, Args) -> - Rpt ! {mnesia_rocksdb, report, R#{time => Time, fmt => Fmt, args => Args}}, +rpt(Rpt, Lbl, Fmt, Args) -> + rpt(Rpt, Lbl, erlang:system_time(millisecond), Fmt, Args). + +rpt(undefined, _, _, _, _) -> ok; +rpt(#{to := Rpt} = R, Lbl, Time, Fmt, Args) -> + Rpt ! {mnesia_rocksdb, report, R#{label => Lbl, time => Time, fmt => Fmt, args => Args}}, ok. +collect_reports(#{ success := SuccessLbl + , failure := ErrorLbl } = Opts) -> + Timeout = maps:get(timeout, Opts, infinity), + receive + {mnesia_rocksdb, report, R} -> + Opts1 = maybe_print(R, Opts), + case R of + #{label := SuccessLbl} -> + collect_ret(ok, Opts1); + #{label := ErrorLbl} -> + collect_ret(error, Opts1); + R -> + collect_reports(Opts1) + end + after Timeout -> + collect_ret(error, Opts) + end. + +maybe_print(progress, #{report := true} = Opts) -> + io:fwrite(".", []), + Opts#{last => progress}; +maybe_print(#{fmt := Fmt, args := Args}, #{report := true} = Opts) -> + case Opts of + #{last := progress} -> + io:fwrite("~n" ++ nl(Fmt), Args); + _ -> + io:fwrite(nl(Fmt), Args) + end, + Opts#{last => fmt}; +maybe_print(_, Opts) -> + Opts. + +nl(S) -> + unicode:characters_to_list([string:chomp(S),"\n"]). + +collect_ret(Ret, Opts) -> + case maps:find(alias, Opts) of + {ok, A} -> + erlang:unalias(A), + Ret; + error -> + Ret + end. + maybe_progress(#{to := To}, C) when C rem 100000 =:= 0 -> To ! {mnesia_rocksdb, report, progress}; maybe_progress(_, _) -> @@ -940,14 +1036,14 @@ do_migrate_tabs(Alias, Tabs, Backend, Rpt, St) -> do_migrate_table(Alias, {Name, OldTRec, TRec0}, Backend, Rpt, St) when is_map(TRec0) -> T0 = erlang:system_time(millisecond), - rpt(Rpt, T0, "Migrate ~p~n", [Name]), + rpt(Rpt, migrating, T0, "Migrate ~p~n", [Name]), TRec = maps:without([encoding, vsn], TRec0), maybe_write_user_props(TRec), {ok, CF, St1} = create_cf_and_migrate(Alias, Name, OldTRec, TRec, Backend, Rpt, St), put_pt(Name, CF), T1 = erlang:system_time(millisecond), - rpt(Rpt, T1, "~nDone (~p)~n", [Name]), + rpt(Rpt, migration_done, T1, "~nDone (~p)~n", [Name]), Time = T1 - T0, io:fwrite("~p migrated, ~p ms~n", [Name, Time]), {{Name, {ok, Time}}, St1}. @@ -1117,6 +1213,28 @@ try_refresh_cf(#{alias := Alias, name := Name, properties := Ps} = Cf, Props, St try_refresh_cf(_, _, _) -> false. +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), + put_pt(Tab, Cf1), + update_cf(Alias, Tab, Cf1, St); + {error, _} -> + error({unknown_cf, {Alias, Tab}}) + 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) -> + UPs0 = maps:get(user_properties, Ps, #{}), + UPs1 = maps:merge(UPs0, maps:from_list(UPs)), + Ps#{user_properties => UPs1}; +update_cf_prop_({K, V}, Ps) -> + Ps#{K => V}. + update_user_properties(Prop, #{properties := Ps} = Cf) -> Key = element(1, Prop), UserProps = case maps:find(user_properties, Ps) of @@ -1534,6 +1652,30 @@ 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; + true -> + {error, invalid_type} + end; + _ -> + %% Start a migration; metadata changed when finalized + start_encoding_migration(Alias, Tab, Opts, Rpt, Backend, St) + end. + start_encoding_migration(Alias, Tab, Encoding0, Rpt, Backend, St) when is_atom(Tab) -> case find_cf(Alias, Tab, Backend, St) of @@ -1557,24 +1699,34 @@ start_encoding_migration(Alias, Tab, Encoding0, Rpt, Backend, St) 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 +start_encoding_migration_(Alias, Tab, Opts, Rpt, OldRef, Backend, St) -> + Props = maps:get(properties, OldRef), + As = maps:get(attributes, Props), + Enc0 = maps:get(encoding, Opts, undefined), + case mnesia_rocksdb_lib:check_encoding(Enc0, As) of {ok, NewEnc} -> case maps:get(encoding, OldRef) of NewEnc -> {error, same_encoding}; _OldEnc -> + Schema = migration_schema_ops(NewEnc, Opts, Props), create_and_start_encoding_mig( - Alias, Tab, NewEnc, Rpt, OldRef, Backend, St) + Alias, Tab, NewEnc, Schema, Rpt, OldRef, Backend, St) end; {error, _} = Err -> Err end. -create_and_start_encoding_mig(Alias, Tab, NewEnc, Rpt, OldRef, +migration_schema_ops(Encoding, Opts, Props) -> + S0 = [{user_properties, [{mrdb_encoding, Encoding}]}], + case Opts of + #{type := Type} when Type =/= map_get(type, Props) -> + [{type, Type}|S0]; + _ -> + S0 + end. + +create_and_start_encoding_mig(Alias, Tab, NewEnc, Schema, Rpt, OldRef, #{db_ref := DbRef} = _Backend, St) -> OldGen = maps:get(cf_gen, OldRef, 0), NewGen = OldGen + 1, @@ -1594,16 +1746,19 @@ create_and_start_encoding_mig(Alias, Tab, NewEnc, Rpt, OldRef, , 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)), + 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} + , schema => Schema , source_gen => OldGen , 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. LiveRef = OldRef#{ migration => NewRef , migration_meta => Meta @@ -1618,7 +1773,7 @@ create_and_start_encoding_mig(Alias, Tab, NewEnc, Rpt, OldRef, {Pid, _MRef} = spawn_monitor( fun() -> encoding_migrator(Alias, Tab, LiveRef2, - NewRef, Rpt) + NewRef, Meta, Rpt) end), rpt(Rpt, "Started encoding migration ~p gen ~p -> ~p (~s)~n", [Tab, OldGen, NewGen, CfName]), @@ -1705,7 +1860,7 @@ resume_encoding_migration_(Alias, Name, LiveRef, Meta, Enc, NewGen, SourceGen, {Pid, _} = spawn_monitor( fun() -> encoding_migrator( - Alias, Name, Live2, NewRef2, + Alias, Name, Live2, NewRef2, Meta1, undefined) end), St#st{migrators = maps:put(Name, Pid, St#st.migrators)} @@ -1716,16 +1871,15 @@ resume_encoding_migration_(Alias, Name, LiveRef, Meta, Enc, NewGen, SourceGen, {Live2, St1}. %% Background copier: walk old CF, install into new if missing (no clobber). -encoding_migrator(Alias, Tab, OldRef, NewRef, Rpt) -> +encoding_migrator(Alias, Tab, OldRef, NewRef, Meta0, 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) }, + 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 -> @@ -1734,12 +1888,12 @@ encoding_migrator(Alias, Tab, OldRef, NewRef, Rpt) -> ok end, write_info(Alias, Tab, encoding_migration, Meta), - rpt(Rpt, "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 -> - ?log(error, "encoding_migrator ~p failed: ~p:~p / ~p", - [Tab, C, R, ST]), + rpt(Rpt, encoding_migrator_failed, "encoding_migrator ~p failed: ~p:~p / ~p", + [Tab, C, R, ST]), error({C, R}) end. @@ -1807,7 +1961,12 @@ do_finalize_encoding_migration(Alias, Tab, _Backend, St) -> Meta = maps:get(migration_meta, LiveRef, #{}), case maps:get(phase, Meta, undefined) of copy_done -> - finalize_encoding_migration_(Alias, Tab, LiveRef, MigRef, St); + update_mnesia_schema( + Tab, + maps:get(schema, Meta), + fun() -> + finalize_encoding_migration_(Alias, Tab, LiveRef, MigRef, St) + end); Phase -> {error, {not_ready, Phase}} end; @@ -1817,14 +1976,65 @@ 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); + #{} -> + {{error, not_migrating}, St}; + error -> + {{error, not_found}, St} + end. + +update_mnesia_schema(Tab, Meta, F) -> + case mnesia_schema:schema_transaction( + fun() -> + update_mnesia_schema_(Tab, Meta, F) + end) of + {atomic, Res} -> + Res; + {aborted, Error} -> + Error + end. + +update_mnesia_schema_(Tab, Meta, F) -> + TidTs = mnesia_schema:get_tid_ts_and_lock(schema, write), + ensure_writable(schema), + Cs = mnesia_schema:incr_version(mnesia_lib:val({Tab, cstruct})), + mnesia_schema:ensure_active(Cs), + List = mnesia_schema:cs2list(Cs), + List1 = lists:foldl( + fun({user_properties, UPs}, Acc) -> + {_, OldUPs} = lists:keyfind(user_properties, 1, Acc), + UPs1 = maps:to_list( + maps:iterator( + maps:merge(maps:from_list(OldUPs), + maps:from_list(UPs)), + ordered)), + lists:keyreplace( + user_properties, 1, Acc, {user_properties, UPs1}); + ({K, V}, Acc) -> + lists:keyreplace(K, 1, Acc, {K, V}) + end, List, Meta), + mnesia_schema:insert_schema_ops(TidTs, [{op, transform, ignore, List1}]), + F(). + +%% This is not exported from mnesia_schema, so copied instead. +ensure_writable(Tab) -> + case mnesia_lib:val({Tab, where_to_write}) of + [] -> + mnesia:abort({read_only, Tab}); + _ -> + ok + end. + %% Cutover: make the versioned target CF the live one and drop the old CF. %% No second copy — the migration target *is* the new generation. 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], + LiveNew = clear_cf_migration( MigRef#{ name => Tab , status => open , encoding => NewEnc @@ -1838,13 +2048,31 @@ finalize_encoding_migration_(Alias, Tab, LiveRef, MigRef, St) -> _ = 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), + try rocksdb:destroy_column_family(DbRef, OldCfH) catch error:_ -> ok end, drop_cached_cf(maps:get(cf_name, LiveRef, data_cf_name(Tab, maps:get(cf_gen, LiveRef, 0))), OldCfH), St2 = St1#st{migrators = maps:remove(Tab, St1#st.migrators)}, {ok, St2}. +abort_encoding_migration_(Alias, Tab, LiveRef, St) -> + case LiveRef of + #{name := Name, migration := MigRef} -> + erase_pt(Name), + #{db_ref := DbRef, cf_handle := CfH} = MigRef, + try rocksdb:drop_column_family(DbRef, CfH) catch error:_ -> ok end, + delete_info(Alias, Tab, encoding_migration), + NewLiveRef = clear_cf_migration(LiveRef), + St1 = update_cf(Alias, Name, NewLiveRef, St), + put_pt(Tab, NewLiveRef), + {ok, St1}; + _ -> + {error, no_migration} + end. + +clear_cf_migration(Cf) -> + maps:without([migration, migration_meta, migration_epoch], Cf). + read_term(Str) -> case erl_scan:string(Str) of {ok, Tokens, _} ->