-
Notifications
You must be signed in to change notification settings - Fork 2.6k
Expand file tree
/
Copy pathhimport_commands.go
More file actions
253 lines (228 loc) · 8.74 KB
/
Copy pathhimport_commands.go
File metadata and controls
253 lines (228 loc) · 8.74 KB
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
package redis
import "context"
// The HIMPORT command family (Redis 8.10+, "hinted hash templates") provides
// fast ingestion of many hashes sharing the same field names. HIMPORT PREPARE
// registers the field names once under a fieldset name, then HIMPORT SET
// creates hashes by sending only the values.
//
// The server scopes a fieldset to the physical connection that prepared it.
// Because go-redis pools connections, the client additionally keeps a
// client-side registry of fieldsets registered through HImportPrepare and
// lazily replays the PREPARE (at most once per connection session) on any
// pooled connection about to execute an HImportSet that references it. See
// himport.go.
//
// The whole HIMPORT surface — the typed methods, the HImport*Cmd types and
// their constructors — is experimental and may be subject to change.
// himportCmder marks HIMPORT commands that participate in client-side
// fieldset tracking. Process paths do a single interface assertion on the
// hot path and inspect the concrete type only for HIMPORT commands.
type himportCmder interface {
Cmder
himportCmd()
}
var (
_ himportCmder = (*HImportPrepareCmd)(nil)
_ himportCmder = (*HImportSetCmd)(nil)
_ himportCmder = (*HImportDiscardCmd)(nil)
_ himportCmder = (*HImportDiscardAllCmd)(nil)
)
// HImportPrepareCmd represents an HIMPORT PREPARE command.
type HImportPrepareCmd struct {
StatusCmd
fieldsetName string
fields []string
// registryVersion and registryEpoch are set only on commands injected by
// the client to replay a registered fieldset onto a connection; on
// success the connection is marked as prepared at this version under
// this discard-all epoch.
registryVersion uint64
registryEpoch uint64
}
func (cmd *HImportPrepareCmd) himportCmd() {}
// NewHImportPrepareCmd returns an HIMPORT PREPARE command.
func NewHImportPrepareCmd(ctx context.Context, fieldsetName string, fields ...string) *HImportPrepareCmd {
args := make([]interface{}, 3+len(fields))
args[0] = "himport"
args[1] = "prepare"
args[2] = fieldsetName
for i, field := range fields {
args[3+i] = field
}
return &HImportPrepareCmd{
StatusCmd: StatusCmd{
baseCmd: baseCmd{
ctx: ctx,
args: args,
cmdType: CmdTypeStatus,
},
},
fieldsetName: fieldsetName,
fields: append([]string(nil), fields...),
}
}
// HImportSetCmd represents an HIMPORT SET command.
type HImportSetCmd struct {
StatusCmd
fieldsetName string
}
func (cmd *HImportSetCmd) himportCmd() {}
// NewHImportSetCmd returns an HIMPORT SET command.
func NewHImportSetCmd(ctx context.Context, key, fieldsetName string, values ...interface{}) *HImportSetCmd {
args := make([]interface{}, 4+len(values))
args[0] = "himport"
args[1] = "set"
args[2] = key
args[3] = fieldsetName
copy(args[4:], values)
cmd := &HImportSetCmd{
StatusCmd: StatusCmd{
baseCmd: baseCmd{
ctx: ctx,
args: args,
cmdType: CmdTypeStatus,
},
},
fieldsetName: fieldsetName,
}
cmd.SetFirstKeyPos(2)
return cmd
}
// HImportDiscardCmd represents an HIMPORT DISCARD command.
type HImportDiscardCmd struct {
IntCmd
fieldsetName string
}
func (cmd *HImportDiscardCmd) himportCmd() {}
// NewHImportDiscardCmd returns an HIMPORT DISCARD command.
func NewHImportDiscardCmd(ctx context.Context, fieldsetName string) *HImportDiscardCmd {
return &HImportDiscardCmd{
IntCmd: IntCmd{
baseCmd: baseCmd{
ctx: ctx,
args: []interface{}{"himport", "discard", fieldsetName},
cmdType: CmdTypeInt,
},
},
fieldsetName: fieldsetName,
}
}
// HImportDiscardAllCmd represents an HIMPORT DISCARDALL command.
type HImportDiscardAllCmd struct {
IntCmd
// registryEpoch is set only on commands injected by the client to wipe a
// session that predates the registry's discard-all epoch; on success the
// connection adopts this epoch.
registryEpoch uint64
}
func (cmd *HImportDiscardAllCmd) himportCmd() {}
// NewHImportDiscardAllCmd returns an HIMPORT DISCARDALL command.
func NewHImportDiscardAllCmd(ctx context.Context) *HImportDiscardAllCmd {
return &HImportDiscardAllCmd{
IntCmd: IntCmd{
baseCmd: baseCmd{
ctx: ctx,
args: []interface{}{"himport", "discardall"},
cmdType: CmdTypeInt,
},
},
}
}
// HImportPrepare registers an ordered list of hash field names under
// fieldsetName for use by subsequent HImportSet calls:
//
// HIMPORT PREPARE fieldset_name field [field ...]
//
// The server keeps the fieldset in the session of the connection that
// executed the command. On pooled clients (Client, Conn, Pipeline, Tx) the
// fieldset is also remembered client-side and the PREPARE is replayed
// lazily — at most once per connection session — on any pooled connection
// about to execute an HImportSet referencing it, so HImportSet works
// transparently across the pool. Preparing an existing fieldset name again
// silently replaces it.
//
// ClusterClient and Ring override this method (see himport_cluster.go): the
// fieldset registers in a registry shared by every node/shard client and the
// PREPARE additionally fans out eagerly to all masters/shards.
//
// Requires Redis 8.10 or newer.
//
// note: the API is experimental and may be subject to change.
func (c cmdable) HImportPrepare(ctx context.Context, fieldsetName string, fields ...string) *StatusCmd {
cmd := NewHImportPrepareCmd(ctx, fieldsetName, fields...)
_ = c(ctx, cmd)
return &cmd.StatusCmd
}
// HImportSet creates or fully replaces the hash at key using the field list
// registered under fieldsetName, pairing values positionally with the
// prepared fields:
//
// HIMPORT SET key fieldset_name value [value ...]
//
// The number of values must equal the fieldset's field count. The resulting
// key is a regular hash readable and writable by all hash commands. If the
// fieldset was registered through HImportPrepare on this client, it is
// prepared automatically on whichever pooled connection executes the command;
// otherwise the fieldset must have been prepared on the executing connection
// or the server replies "ERR no such fieldset".
//
// "no such fieldset" never surfaces for a registered fieldset: a
// single-command HImportSet whose connection lost its session state (e.g.
// RESET) is transparently re-prepared and retried once — the failure also
// stales every other connection's prepared flag, so the retry re-prepares
// wherever it lands, and this recovery attempt is granted even when retries
// are disabled (MaxRetries -1). In pipelines the failed HImportSets — and
// only those — are re-prepared and re-issued once on the same connection
// (HIMPORT SET is a full replace, so the re-execution is idempotent and no
// other command of the batch runs again). Inside transactions the error does
// surface after EXEC — an executed transaction cannot be partially re-run —
// but the prepared flags are invalidated, so retrying the transaction
// succeeds.
//
// Requires Redis 8.10 or newer.
//
// note: the API is experimental and may be subject to change.
func (c cmdable) HImportSet(ctx context.Context, key, fieldsetName string, values ...interface{}) *StatusCmd {
cmd := NewHImportSetCmd(ctx, key, fieldsetName, values...)
_ = c(ctx, cmd)
return &cmd.StatusCmd
}
// HImportDiscard removes fieldsetName from the executing connection's session
// and from the client-side registry, stopping further automatic replay:
//
// HIMPORT DISCARD fieldset_name
//
// It returns 1 if the fieldset was registered on this client and is now
// removed, 0 otherwise (for names never registered through the managed API,
// the executing connection's session reply passes through unchanged). Pooled
// connections whose sessions still hold the fieldset replay the DISCARD
// before their next HIMPORT command, so a subsequent HImportSet fails with
// "no such fieldset" on every connection, exactly as on a single connection.
// Hashes already created through the fieldset are not affected.
//
// Requires Redis 8.10 or newer.
//
// note: the API is experimental and may be subject to change.
func (c cmdable) HImportDiscard(ctx context.Context, fieldsetName string) *IntCmd {
cmd := NewHImportDiscardCmd(ctx, fieldsetName)
_ = c(ctx, cmd)
return &cmd.IntCmd
}
// HImportDiscardAll removes all fieldsets from the executing connection's
// session and clears the client-side registry:
//
// HIMPORT DISCARDALL
//
// It returns the number of fieldsets removed from the client-side registry
// (when none were registered, the executing connection's session count
// passes through). Other pooled connections whose sessions were prepared
// earlier replay HIMPORT DISCARDALL before their next HIMPORT command.
//
// Requires Redis 8.10 or newer.
//
// note: the API is experimental and may be subject to change.
func (c cmdable) HImportDiscardAll(ctx context.Context) *IntCmd {
cmd := NewHImportDiscardAllCmd(ctx)
_ = c(ctx, cmd)
return &cmd.IntCmd
}