forked from cockroachdb/cockroach
/
writebatch.go
91 lines (78 loc) · 3.14 KB
/
writebatch.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
// Copyright 2016 The Cockroach Authors.
//
// Licensed under the Cockroach Community Licence (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://github.com/cockroachdb/cockroach/blob/master/pkg/ccl/LICENSE
package storageccl
import (
"errors"
"fmt"
"golang.org/x/net/context"
"github.com/cockroachdb/cockroach/pkg/ccl/storageccl/engineccl"
"github.com/cockroachdb/cockroach/pkg/roachpb"
"github.com/cockroachdb/cockroach/pkg/storage"
"github.com/cockroachdb/cockroach/pkg/storage/engine"
"github.com/cockroachdb/cockroach/pkg/util/tracing"
)
func init() {
storage.SetWriteBatchCmd(storage.Command{
DeclareKeys: storage.DefaultDeclareKeys,
Eval: evalWriteBatch,
})
}
// evalWriteBatch applies the operations encoded in a BatchRepr. Any existing
// data in the affected keyrange is first cleared (not tombstoned), which makes
// this command idempotent.
func evalWriteBatch(
ctx context.Context, batch engine.ReadWriter, cArgs storage.CommandArgs, _ roachpb.Response,
) (storage.EvalResult, error) {
if !storage.ProposerEvaluatedKVEnabled() {
// To reduce the footprint of things that have ever been downstream of
// raft, forbid this command from running without proposer evaluated kv.
panic("command WriteBatch is not allowed without proposer evaluated KV")
}
args := cArgs.Args.(*roachpb.WriteBatchRequest)
h := cArgs.Header
ms := cArgs.Stats
_, span := tracing.ChildSpan(ctx, fmt.Sprintf("WriteBatch [%s,%s)", args.Key, args.EndKey))
defer tracing.FinishSpan(span)
// We can't use the normal RangeKeyMismatchError mechanism for dealing with
// splits because args.Data should stay an opaque blob to DistSender.
if args.DataSpan.Key.Compare(args.Key) < 0 || args.DataSpan.EndKey.Compare(args.EndKey) > 0 {
// TODO(dan): Add a new field in roachpb.Error, so the client can catch
// this and retry.
return storage.EvalResult{}, errors.New("data spans multiple ranges")
}
mvccStartKey := engine.MVCCKey{Key: args.Key}
mvccEndKey := engine.MVCCKey{Key: args.EndKey}
// Verify that the keys in the batch are within the range specified by the
// request header.
msBatch, err := engineccl.VerifyBatchRepr(args.Data, mvccStartKey, mvccEndKey, h.Timestamp.WallTime)
if err != nil {
return storage.EvalResult{}, err
}
ms.Add(msBatch)
// Check if there was data in the affected keyrange. If so, delete it (and
// adjust the MVCCStats) before applying the WriteBatch data.
iter := batch.NewIterator(false)
defer iter.Close()
iter.Seek(mvccStartKey)
if iter.Valid() && iter.Key().Less(mvccEndKey) {
existingStats, err := iter.ComputeStats(mvccStartKey, mvccEndKey, h.Timestamp.WallTime)
if err != nil {
return storage.EvalResult{}, err
}
// TODO(dan): Ideally, this would use `batch.ClearRange` but it doesn't
// yet work with read-write batches.
if err := batch.ClearIterRange(iter, mvccStartKey, mvccEndKey); err != nil {
return storage.EvalResult{}, err
}
ms.Subtract(existingStats)
}
if err := batch.ApplyBatchRepr(args.Data, false /* !sync */); err != nil {
return storage.EvalResult{}, err
}
return storage.EvalResult{}, nil
}