Skip to content

Commit accb87e

Browse files
CASSGO-26 consistency serial was added
The user should be able to set consistency to SERIAL or LOCAL_SERIAL for Paxos reads, but the previous implementation doesn't support such a feature. patch by Mykyta Oleksiienko; reviewed by João Reis for [CASSGO-26](https://issues.apache.org/jira/browse/CASSGO-26)
1 parent 7b7e6af commit accb87e

File tree

4 files changed

+37
-43
lines changed

4 files changed

+37
-43
lines changed

CHANGELOG.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
88

99
### Added
1010

11+
- Allow SERIAL and LOCAL_SERIAL on SELECT statements [CASSGO-26](https://issues.apache.org/jira/browse/CASSGO-26)
12+
1113
### Changed
1214

1315
- Don't restrict server authenticator unless PasswordAuthentictor.AllowedAuthenticators is provided (CASSGO-19)

cluster.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -150,7 +150,7 @@ type ClusterConfig struct {
150150

151151
// Consistency for the serial part of queries, values can be either SERIAL or LOCAL_SERIAL.
152152
// Default: unset
153-
SerialConsistency SerialConsistency
153+
SerialConsistency Consistency
154154

155155
// SslOpts configures TLS use when HostDialer is not set.
156156
// SslOpts is ignored if HostDialer is set.

frame.go

Lines changed: 20 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -192,6 +192,9 @@ const (
192192

193193
type Consistency uint16
194194

195+
// SerialConsistency is deprecated. Use Consistency instead.
196+
type SerialConsistency = Consistency
197+
195198
const (
196199
Any Consistency = 0x00
197200
One Consistency = 0x01
@@ -202,6 +205,8 @@ const (
202205
LocalQuorum Consistency = 0x06
203206
EachQuorum Consistency = 0x07
204207
LocalOne Consistency = 0x0A
208+
Serial Consistency = 0x08
209+
LocalSerial Consistency = 0x09
205210
)
206211

207212
func (c Consistency) String() string {
@@ -224,6 +229,10 @@ func (c Consistency) String() string {
224229
return "EACH_QUORUM"
225230
case LocalOne:
226231
return "LOCAL_ONE"
232+
case Serial:
233+
return "SERIAL"
234+
case LocalSerial:
235+
return "LOCAL_SERIAL"
227236
default:
228237
return fmt.Sprintf("UNKNOWN_CONS_0x%x", uint16(c))
229238
}
@@ -253,13 +262,21 @@ func (c *Consistency) UnmarshalText(text []byte) error {
253262
*c = EachQuorum
254263
case "LOCAL_ONE":
255264
*c = LocalOne
265+
case "SERIAL":
266+
*c = Serial
267+
case "LOCAL_SERIAL":
268+
*c = LocalSerial
256269
default:
257270
return fmt.Errorf("invalid consistency %q", string(text))
258271
}
259272

260273
return nil
261274
}
262275

276+
func (c Consistency) IsSerial() bool {
277+
return c == Serial || c == LocalSerial
278+
279+
}
263280
func ParseConsistency(s string) Consistency {
264281
var c Consistency
265282
if err := c.UnmarshalText([]byte(strings.ToUpper(s))); err != nil {
@@ -286,41 +303,6 @@ func MustParseConsistency(s string) (Consistency, error) {
286303
return c, nil
287304
}
288305

289-
type SerialConsistency uint16
290-
291-
const (
292-
Serial SerialConsistency = 0x08
293-
LocalSerial SerialConsistency = 0x09
294-
)
295-
296-
func (s SerialConsistency) String() string {
297-
switch s {
298-
case Serial:
299-
return "SERIAL"
300-
case LocalSerial:
301-
return "LOCAL_SERIAL"
302-
default:
303-
return fmt.Sprintf("UNKNOWN_SERIAL_CONS_0x%x", uint16(s))
304-
}
305-
}
306-
307-
func (s SerialConsistency) MarshalText() (text []byte, err error) {
308-
return []byte(s.String()), nil
309-
}
310-
311-
func (s *SerialConsistency) UnmarshalText(text []byte) error {
312-
switch string(text) {
313-
case "SERIAL":
314-
*s = Serial
315-
case "LOCAL_SERIAL":
316-
*s = LocalSerial
317-
default:
318-
return fmt.Errorf("invalid consistency %q", string(text))
319-
}
320-
321-
return nil
322-
}
323-
324306
const (
325307
apacheCassandraTypePrefix = "org.apache.cassandra.db.marshal."
326308
)
@@ -1452,7 +1434,7 @@ type queryParams struct {
14521434
values []queryValues
14531435
pageSize int
14541436
pagingState []byte
1455-
serialConsistency SerialConsistency
1437+
serialConsistency Consistency
14561438
// v3+
14571439
defaultTimestamp bool
14581440
defaultTimestampValue int64
@@ -1541,7 +1523,7 @@ func (f *framer) writeQueryParams(opts *queryParams) {
15411523
}
15421524

15431525
if opts.serialConsistency > 0 {
1544-
f.writeConsistency(Consistency(opts.serialConsistency))
1526+
f.writeConsistency(opts.serialConsistency)
15451527
}
15461528

15471529
if f.proto > protoVersion2 && opts.defaultTimestamp {
@@ -1653,7 +1635,7 @@ type writeBatchFrame struct {
16531635
consistency Consistency
16541636

16551637
// v3+
1656-
serialConsistency SerialConsistency
1638+
serialConsistency Consistency
16571639
defaultTimestamp bool
16581640
defaultTimestampValue int64
16591641

session.go

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,10 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
144144
return nil, errors.New("Can't use both Authenticator and AuthProvider in cluster config.")
145145
}
146146

147+
if cfg.SerialConsistency > 0 && !cfg.SerialConsistency.IsSerial() {
148+
return nil, fmt.Errorf("the default SerialConsistency level is not allowed to be anything else but SERIAL or LOCAL_SERIAL. Recived value: %v", cfg.SerialConsistency)
149+
}
150+
147151
// TODO: we should take a context in here at some point
148152
ctx, cancel := context.WithCancel(context.TODO())
149153

@@ -915,7 +919,7 @@ type Query struct {
915919
rt RetryPolicy
916920
spec SpeculativeExecutionPolicy
917921
binding func(q *QueryInfo) ([]interface{}, error)
918-
serialCons SerialConsistency
922+
serialCons Consistency
919923
defaultTimestamp bool
920924
defaultTimestampValue int64
921925
disableSkipMetadata bool
@@ -1264,7 +1268,10 @@ func (q *Query) Bind(v ...interface{}) *Query {
12641268
// either SERIAL or LOCAL_SERIAL and if not present, it defaults to
12651269
// SERIAL. This option will be ignored for anything else that a
12661270
// conditional update/insert.
1267-
func (q *Query) SerialConsistency(cons SerialConsistency) *Query {
1271+
func (q *Query) SerialConsistency(cons Consistency) *Query {
1272+
if !cons.IsSerial() {
1273+
panic("Serial consistency can only be SERIAL or LOCAL_SERIAL got " + cons.String())
1274+
}
12681275
q.serialCons = cons
12691276
return q
12701277
}
@@ -1735,7 +1742,7 @@ type Batch struct {
17351742
trace Tracer
17361743
observer BatchObserver
17371744
session *Session
1738-
serialCons SerialConsistency
1745+
serialCons Consistency
17391746
defaultTimestamp bool
17401747
defaultTimestampValue int64
17411748
context context.Context
@@ -1914,7 +1921,10 @@ func (b *Batch) Size() int {
19141921
// conditional update/insert.
19151922
//
19161923
// Only available for protocol 3 and above
1917-
func (b *Batch) SerialConsistency(cons SerialConsistency) *Batch {
1924+
func (b *Batch) SerialConsistency(cons Consistency) *Batch {
1925+
if !cons.IsSerial() {
1926+
panic("Serial consistency can only be SERIAL or LOCAL_SERIAL got " + cons.String())
1927+
}
19181928
b.serialCons = cons
19191929
return b
19201930
}

0 commit comments

Comments
 (0)