anchor.diff:      RObj = riak_object:newts(
                Bucket, PK, Obj,
                dict:from_list([{?MD_DDL_VERSION, ?DDL_VERSION}])),
  -    {RObj, LK}.
  -
  +    {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).
  +%%%%%%%%
   
   
   
  
positive.code:                   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};
                        _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.
  
  
  
  
negative.code:     end.
  
  put_data_to_partitions(Data, Bucket, BucketProps, DDL, Mod) ->
      DDL = Mod:get_ddl(),
      BucketProps = riak_core_bucket:get_bucket(Bucket),
      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)),
  
      SendFullBatches = riak_core_capability:get({riak_kv, w1c_batch_vnode}, false),
      %% Default to 1MB for a max batch size to not overwhelm disterl
      CappedBatchSize = app_helper:get_env(riak_kv, timeseries_max_batch_size,
                                           1024 * 1024),
      EncodeFn =
          fun(O) -> riak_object:to_binary(v1, O, msgpack) end,
  
  