anchor.diff:                    case riak_kv_w1c_worker:validate_options(
                            NVal, Preflist, [], BucketProps) of
                         {ok, W, PW} ->
  -                          {Ids, Errs} =
  -                              lists:foldl(
  -                                fun(Record, {PartReqIds, PartErrors}) ->
  -                                        {RObj, LK} =
  -                                            build_object(Bucket, Mod, DDL,
  -                                                         Record, DocIdx),
  -
  -                                        {ok, ReqId} =
  -                                            riak_kv_w1c_worker:async_put(
  -                                              RObj, W, PW, Bucket, NVal, LK,
  -                                              EncodeFn, Preflist),
  -                                        {[ReqId | PartReqIds], PartErrors}
  -                                end,
  -                                {[], 0}, Records),
  -                          {GlobalReqIds ++ Ids, GlobalErrorsCnt + Errs};
  +                          DataForVnode = pick_batch_option(SendFullBatches,
  +                                                           CappedBatchSize,
  +                                                           Records,
  +                                                           termsize(hd(Records)),
  +                                                           length(Records)),
  +                          Ids =
  +                              invoke_async_put(fun(Record) ->
  +                                                       build_object(Bucket, Mod, DDL,
  +                                                                    Record, DocIdx)
  +                                               end,
  +                                               fun(RObj, LK) ->
  +                                                       riak_kv_w1c_worker:async_put(
  +                                                         RObj, W, PW, Bucket, NVal, LK,
  +                                                         EncodeFn, Preflist)
  +                                               end,
  +                                               fun(RObjs) ->
  +                                                       riak_kv_w1c_worker:ts_batch_put(
  +                                                         RObjs, W, PW, Bucket, NVal,
  +                                                         EncodeFn, Preflist)
  +                                               end,
  +                                               DataForVnode),
  +                          {GlobalReqIds ++ Ids, GlobalErrorsCnt};
                         _Error ->
                             {GlobalReqIds, GlobalErrorsCnt + length(Records)}
                     end
             end,
             {[], 0}, PreflistData),
       Responses = riak_kv_w1c_worker:async_put_replies(ReqIds, []),
  -    _NErrors =
  -        length(
  -          lists:filter(
  -            fun({error, _}) -> true;
  -               (_) -> false
  -            end, Responses)) + FailReqs.
  +    length(lists:filter(fun({error, _}) -> true;
  +                           (_) -> false
  +                        end, Responses)) + FailReqs.
   
   
   
  
positive.code:     end.
  
  put_data_to_partitions(Data, Bucket, BucketProps, DDL, Mod) ->
      NVal = proplists:get_value(n_val, BucketProps),
      PartitionedData = partition_data(Data, Bucket, BucketProps, DDL, Mod),
      PreflistData = add_preflists(PartitionedData, NVal,
                                   riak_core_node_watcher:nodes(riak_kv)),
      EncodeFn =
          fun(O) -> riak_object:to_binary(v1, O, msgpack) end,
  
  
negative.code:     RObj = riak_object:newts(
               Bucket, PK, Obj,
               dict:from_list([{?MD_DDL_VERSION, ?DDL_VERSION}])),
      {LK, RObj}.
  
  %%%%%%%%
  %% Utility functions for batch delivery of records
  termsize(Term) ->
      size(term_to_binary(Term)).
  
  pick_batch_option(_, _, Records, _, 1) ->
      {individual, Records};
  pick_batch_option(true, MaxBatch, Records, SampleSize, _NumRecs) ->
      {batches, create_batches(Records,
                               estimated_row_count(SampleSize, MaxBatch))};
  pick_batch_option(false, _, Records, _, _) ->
      {individual, Records}.
  
  estimated_row_count(SampleRowSize, MaxBatchSize) ->
      %% Assume some rows will be larger, so introduce a fudge factor of
      %% roughly 10 percent.
      RowSizeFudged = (SampleRowSize * 10) div 9,
      MaxBatchSize div RowSizeFudged.
  
  create_batches(Rows, MaxSize) ->
      create_batches(Rows, MaxSize, []).
  
  create_batches([], _MaxSize, Accum) ->
      Accum;
  create_batches(Rows, MaxSize, Accum) when length(Rows) < MaxSize ->
      [Rows|Accum];
  create_batches(Rows, MaxSize, Accum) ->
      {First, Rest} = lists:split(MaxSize, Rows),
      create_batches(Rest, MaxSize, [First|Accum]).
  
  %% Returns a tuple with a list of request IDs and an error tally
  invoke_async_put(BuildRObjFun, AsyncPutFun, _BatchPutFun, {individual, Records}) ->
      lists:map(fun(Record) ->
                        {LK, RObj} = BuildRObjFun(Record),
                        {ok, ReqId} = AsyncPutFun(RObj, LK),
                        ReqId
                  end,
                Records);
  invoke_async_put(BuildRObjFun, _AsyncPutFun, BatchPutFun, {batches, Batches}) ->
      lists:map(fun(Batch) ->
                        RObjs = lists:map(BuildRObjFun, Batch),
                        {ok, ReqId} = BatchPutFun(RObjs),
                        ReqId
                  end,
                Batches).
  %%%%%%%%
  
  
  
  