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:     RObj = riak_object:newts(
               Bucket, PK, Obj,
               dict:from_list([{?MD_DDL_VERSION, ?DDL_VERSION}])),
      {RObj, LK}.
  
  
  
  
  
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,
  
  