1
0
mirror of https://github.com/uptrace/go-clickhouse.git synced 2025-06-27 00:21:13 +02:00

feat: split ch.In into ch.List, ch.Array, and ch.In

This commit is contained in:
Vladimir Mihailenco
2023-05-30 15:42:52 +03:00
parent c0e67d2ec4
commit 13e4911570
4 changed files with 93 additions and 56 deletions

View File

@ -70,34 +70,89 @@ func isBadConn(err error, allowTimeout bool) bool {
//------------------------------------------------------------------------------ //------------------------------------------------------------------------------
type ListValues struct {
slice any
}
var _ chschema.QueryAppender = ListValues{}
func List(slice any) ListValues {
return ListValues{
slice: slice,
}
}
func (in ListValues) AppendQuery(fmter chschema.Formatter, b []byte) (_ []byte, err error) {
v := reflect.ValueOf(in.slice)
if v.Kind() != reflect.Slice {
return nil, fmt.Errorf("ch: In(non-slice %T)", in.slice)
}
b = appendList(fmter, b, v)
return b, nil
}
//------------------------------------------------------------------------------
type InValues struct { type InValues struct {
slice reflect.Value slice any
err error
} }
var _ chschema.QueryAppender = InValues{} var _ chschema.QueryAppender = InValues{}
func In(slice any) InValues { func In(slice any) InValues {
v := reflect.ValueOf(slice)
if v.Kind() != reflect.Slice {
return InValues{
err: fmt.Errorf("ch: In(non-slice %T)", slice),
}
}
return InValues{ return InValues{
slice: v, slice: slice,
} }
} }
func (in InValues) AppendQuery(fmter chschema.Formatter, b []byte) (_ []byte, err error) { func (in InValues) AppendQuery(fmter chschema.Formatter, b []byte) (_ []byte, err error) {
if in.err != nil { v := reflect.ValueOf(in.slice)
return nil, in.err if v.Kind() != reflect.Slice {
return nil, fmt.Errorf("ch: In(non-slice %T)", in.slice)
} }
return appendIn(fmter, b, in.slice), nil
b = append(b, '(')
b = appendList(fmter, b, v)
b = append(b, ')')
return b, nil
} }
func appendIn(fmter chschema.Formatter, b []byte, slice reflect.Value) []byte { //------------------------------------------------------------------------------
type ArrayValues struct {
slice any
}
var _ chschema.QueryAppender = ArrayValues{}
func Array(slice any) ArrayValues {
return ArrayValues{
slice: slice,
}
}
func (in ArrayValues) AppendQuery(fmter chschema.Formatter, b []byte) (_ []byte, err error) {
v := reflect.ValueOf(in.slice)
if v.Kind() != reflect.Slice {
return nil, fmt.Errorf("ch: Array(non-slice %T)", in.slice)
}
b = append(b, '[')
b = appendList(fmter, b, v)
b = append(b, ']')
return b, nil
}
//------------------------------------------------------------------------------
func appendList(fmter chschema.Formatter, b []byte, slice reflect.Value) []byte {
sliceLen := slice.Len() sliceLen := slice.Len()
if sliceLen == 0 {
return append(b, "NULL"...)
}
for i := 0; i < sliceLen; i++ { for i := 0; i < sliceLen; i++ {
if i > 0 { if i > 0 {
b = append(b, ", "...) b = append(b, ", "...)
@ -108,13 +163,7 @@ func appendIn(fmter chschema.Formatter, b []byte, slice reflect.Value) []byte {
elem = elem.Elem() elem = elem.Elem()
} }
if elem.Kind() == reflect.Slice { b = chschema.AppendValue(fmter, b, elem)
b = append(b, '(')
b = appendIn(fmter, b, elem)
b = append(b, ')')
} else {
b = chschema.AppendValue(fmter, b, elem)
}
} }
return b return b
} }

View File

@ -76,9 +76,9 @@ func defaultConfig() *Config {
ReadTimeout: 30 * time.Second, ReadTimeout: 30 * time.Second,
WriteTimeout: 10 * time.Second, WriteTimeout: 10 * time.Second,
MaxRetries: 2, MaxRetries: 3,
MinRetryBackoff: 500 * time.Millisecond, MinRetryBackoff: time.Millisecond,
MaxRetryBackoff: time.Second, MaxRetryBackoff: 3 * time.Second,
} }
return conf return conf
} }

View File

@ -272,9 +272,8 @@ func (db *DB) exec(ctx context.Context, query string) (*result, error) {
var lastErr error var lastErr error
for attempt := 0; attempt <= db.conf.MaxRetries; attempt++ { for attempt := 0; attempt <= db.conf.MaxRetries; attempt++ {
if attempt > 0 { if attempt > 0 {
lastErr = internal.Sleep(ctx, db.retryBackoff(attempt-1)) if err := internal.Sleep(ctx, db.retryBackoff()); err != nil {
if lastErr != nil { return nil, err
break
} }
} }
@ -338,9 +337,8 @@ func (db *DB) query(ctx context.Context, query string) (*blockIter, error) {
for attempt := 0; attempt <= db.conf.MaxRetries; attempt++ { for attempt := 0; attempt <= db.conf.MaxRetries; attempt++ {
if attempt > 0 { if attempt > 0 {
lastErr = internal.Sleep(ctx, db.retryBackoff(attempt-1)) if err := internal.Sleep(ctx, db.retryBackoff()); err != nil {
if lastErr != nil { return nil, err
break
} }
} }
@ -379,9 +377,8 @@ func (db *DB) insert(
for attempt := 0; attempt <= db.conf.MaxRetries; attempt++ { for attempt := 0; attempt <= db.conf.MaxRetries; attempt++ {
if attempt > 0 { if attempt > 0 {
lastErr = internal.Sleep(ctx, db.retryBackoff(attempt-1)) if err := internal.Sleep(ctx, db.retryBackoff()); err != nil {
if lastErr != nil { return nil, err
break
} }
} }
@ -489,22 +486,28 @@ func (db *DB) WithFormatter(fmter chschema.Formatter) *DB {
func (db *DB) shouldRetry(err error) bool { func (db *DB) shouldRetry(err error) bool {
switch err { switch err {
case driver.ErrBadConn:
return true
case nil, context.Canceled, context.DeadlineExceeded: case nil, context.Canceled, context.DeadlineExceeded:
return false return false
case driver.ErrBadConn:
return true
} }
if err, ok := err.(*Error); ok { if err, ok := err.(*Error); ok {
// https://github.com/ClickHouse/ClickHouse/blob/master/src/Common/ErrorCodes.cpp // https://github.com/ClickHouse/ClickHouse/blob/master/src/Common/ErrorCodes.cpp
const ( const (
timeoutExceeded = 159 timeoutExceeded = 159
tooSlow = 160
tooManySimultaneousQueries = 202 tooManySimultaneousQueries = 202
memoryLimitExceeded = 241 memoryLimitExceeded = 241
cannotDecompress = 271
) )
switch err.Code { switch err.Code {
case timeoutExceeded, tooManySimultaneousQueries, memoryLimitExceeded: case timeoutExceeded,
tooSlow,
tooManySimultaneousQueries,
memoryLimitExceeded,
cannotDecompress:
return true return true
} }
} }
@ -512,9 +515,8 @@ func (db *DB) shouldRetry(err error) bool {
return false return false
} }
func (db *DB) retryBackoff(attempt int) time.Duration { func (db *DB) retryBackoff() time.Duration {
return internal.RetryBackoff( return internal.RetryBackoff(db.conf.MinRetryBackoff, db.conf.MaxRetryBackoff)
attempt, db.conf.MinRetryBackoff, db.conf.MaxRetryBackoff)
} }
func (db *DB) FormatQuery(query string, args ...any) string { func (db *DB) FormatQuery(query string, args ...any) string {

View File

@ -79,24 +79,10 @@ func MakeSliceNextElemFunc(v reflect.Value) func() reflect.Value {
} }
} }
func RetryBackoff(retry int, minBackoff, maxBackoff time.Duration) time.Duration { func RetryBackoff(minBackoff, maxBackoff time.Duration) time.Duration {
if retry < 0 { backoff := minBackoff + time.Duration(rand.Int63n(int64(maxBackoff)))
panic("not reached") if backoff > maxBackoff {
}
if minBackoff == 0 {
return 0
}
d := minBackoff << uint(retry)
if d < minBackoff {
return maxBackoff return maxBackoff
} }
return backoff
d = minBackoff + time.Duration(rand.Int63n(int64(d)))
if d > maxBackoff || d < minBackoff {
d = maxBackoff
}
return d
} }