anchor.diff:      end.
   
   %% @private
  -check_symlink(LinkFile, DeleteExisting) ->
  -    case file:read_link(LinkFile) of
  -        {ok, DataFile} ->
  -            case DeleteExisting of
  -                true ->
  -                    file:delete(LinkFile),
  -                    make_symlink(LinkFile);
  -                false ->
  -                    {ok, DataFile}
  -            end;
  -        {error, enoent} ->
  -            make_symlink(LinkFile);
  -        Error ->
  -            Error
  -    end.
  -
  -%% @private
  -make_symlink(LinkFile) ->
  +get_data_dir(DataRoot, PartitionI) ->
  +    Partition = integer_to_list(PartitionI),
  +    PartitionPath = filename:join([DataRoot, Partition]),
  +    case filelib:is_dir(PartitionPath) of
  +        true ->
  +            {ok, Partition};
  +        false ->
  +            %% Check for any existing directories for the partition
  +            %% and select the most recent as the active data directory
  +            %% if any exist.
  +            case filelib:wildcard(PartitionPath ++ "-*") of
  +                [] ->
  +                    make_data_dir(PartitionPath);
  +                PartitionDirs ->
  +                    [DataDir | RestPartitionDirs] =
  +                        lists:reverse(PartitionDirs),
  +                    log_unused_partition_dirs(Partition,
  +                                              RestPartitionDirs),
  +                    {ok, filename:basename(DataDir)}
  +            end
  +    end.
  +
  +%% @private
  +log_unused_partition_dirs(Partition, PartitionDirs) ->
  +    case PartitionDirs of
  +        [] ->
  +            ok;
  +        _ ->
  +            %% Inform the user in case they want to do some cleanup.
  +            lager:notice("Unused data directories exist for partition ~p: ~p",
  +                       [Partition, PartitionDirs])
  +    end.
  +
  +%% @private
  +make_data_dir(PartitionFile) ->
       {MegaSecs, Secs, MicroSecs} = erlang:now(),
  -    LinkAbsPath = filename:absname(LinkFile),
  -    DataFile = [filename:basename(LinkFile),
  -                "-",
  -                integer_to_list(MegaSecs),
  -                integer_to_list(Secs),
  -                integer_to_list(MicroSecs)],
  -    case filelib:ensure_dir(LinkAbsPath) of
  +    AbsPath = filename:absname(PartitionFile),
  +    DataDir = [filename:basename(PartitionFile),
  +               "-",
  +               integer_to_list(MegaSecs),
  +               integer_to_list(Secs),
  +               integer_to_list(MicroSecs)],
  +    case filelib:ensure_dir([AbsPath, DataDir]) of
           ok ->
  -            case file:make_symlink(DataFile,
  -                                   LinkAbsPath) of
  -                ok ->
  -                    {ok, DataFile};
  -                {error, Reason} ->
  -                    lager:error("Failed to create bitcask dir ~s: ~p",
  -                                [DataFile, Reason]),
  -                    {error, Reason}
  -            end;
  -        {error, Reason1} ->
  +            {ok, DataDir};
  +        {error, Reason} ->
               lager:error("Failed to create bitcask dir ~s: ~p",
  -                        [DataFile, Reason1]),
  -            {error, Reason1}
  -    end.
  -
  -%% @private
  -drop_data_cleanup(DataRoot, LinkFile, DataFile) ->
  +                        [DataDir, Reason]),
  +            {error, Reason}
  +    end.
  +
  +%% @private
  +drop_data_cleanup(DataRoot, Partition, DataDir) ->
       fun() ->
               %% List all the directories in data root
               case file:list_dir(DataRoot) of
                   {ok, Dirs} ->
  -                    LinkBase = filename:basename(LinkFile),
  +                    PartitionStr = integer_to_list(Partition),
                       %% Delete the contents of each directory and
                       %% the directory itself excluding the
  -                    %% current data directory and the symlink
  -                    %% pointing to it.
  +                    %% current data directory.
                       [data_directory_cleanup(filename:join(DataRoot, Dir)) ||
                           Dir <- Dirs,
  -                        Dir /= lists:flatten(DataFile),
  -                        Dir /= LinkBase,
  -                        string:left(Dir, length(LinkBase) + 1) =:= (LinkBase ++ "-")];
  +                        Dir /= lists:flatten(DataDir),
  +                        string:left(Dir, length(PartitionStr) + 1) =:= (PartitionStr ++ "-")];
                   {error, _} ->
                       ignore
               end
  
positive.code:                    [{atom(), term()}],
                     state()) -> {ok, any()} | {async, fun()} | {error, term()}.
  fold_objects(FoldObjectsFun, Acc, Opts, #state{opts=BitcaskOpts,
                                                 data_file=DataFile,
                                                 root=DataRoot,
                                                 async_folds=true}) ->
      Bucket =  proplists:get_value(bucket, Opts),
  
negative.code: -define(CAPABILITIES, [async_fold]).
  
  -record(state, {ref :: reference(),
                  data_dir :: string(),
                  opts :: [{atom(), term()}],
                  partition :: integer(),
                  root :: string(),
                  async_folds :: boolean()}).
  
  