anchor.diff:                     [],
                      state()) -> {ok, any()} | {async, fun()} | {error, term()}.
   fold_buckets(FoldBucketsFun, Acc, _Opts, #state{opts=BitcaskOpts,
  -                                                data_file=DataFile,
  +                                                data_dir=DataFile,
                                                   root=DataRoot,
                                                   async_folds=true}) ->
       FoldFun = fold_buckets_fun(FoldBucketsFun),
  
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:     end.
  
  %% @private
  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(),
      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 ->
              {ok, DataDir};
          {error, Reason} ->
              lager:error("Failed to create bitcask dir ~s: ~p",
                          [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} ->
                      PartitionStr = integer_to_list(Partition),
                      %% Delete the contents of each directory and
                      %% the directory itself excluding the
                      %% current data directory.
                      [data_directory_cleanup(filename:join(DataRoot, Dir)) ||
                          Dir <- Dirs,
                          Dir /= lists:flatten(DataDir),
                          string:left(Dir, length(PartitionStr) + 1) =:= (PartitionStr ++ "-")];
                  {error, _} ->
                      ignore
              end
  