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(Rpt, Lbl, Fmt, Args) ->
-rpt(#{to := Rpt} = R, Time, Fmt, Args) -> 
+    rpt(Rpt, Lbl, erlang:system_time(millisecond), Fmt, Args).
-    Rpt ! {mnesia_rocksdb, report, R#{time => Time, fmt => Fmt, args => 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) ->
+start_encoding_migration_(Alias, Tab, Opts, Rpt, OldRef, Backend, St) ->
-    As = maps:get(attributes,
+    Props = maps:get(properties, OldRef),
-                  maps:get(properties, OldRef, #{}),
+    As = maps:get(attributes, Props),
-                  [key, val]),
+    Enc0 = maps:get(encoding, Opts, undefined),
-    case mnesia_rocksdb_lib:check_encoding(Encoding0, As) of
+    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(
+            NewRef1 = update_cf_props(Schema, check_version_and_encoding(NewRef0b)),
-                        {mrdb_encoding, NewEnc},
+            %% NewRef1 = update_user_properties(
-                        check_version_and_encoding(NewRef0b)),
+            %%             {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
+        Meta = Meta0#{ phase => copy_done
-                , target => #{encoding => maps:get(encoding, NewRef)}
+                     , cursor => '$end'
-                , cursor => '$end'
+                     , epoch => 1
-                , epoch => 1
+                     , copied => N
-                , copied => N
+                     , finished_at => erlang:system_time(millisecond) },
-                , 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",
+            rpt(Rpt, encoding_migrator_failed, "encoding_migrator ~p failed: ~p:~p / ~p",
-                 [Tab, C, R, ST]),
+                [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(
+    LiveNew = clear_cf_migration(
-                [migration, migration_meta, migration_epoch],
                 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, _} ->