diff --git a/bindings.go b/bindings.go index 95a512b3f..10c47f792 100644 --- a/bindings.go +++ b/bindings.go @@ -99,7 +99,7 @@ func packItem(ns *reindexerNamespace, item any, json []byte, ser *cjson.Serializ } if t != ns.rtype { if ns.rtype.Name() != t.Name() || ns.rtype.PkgPath() != t.PkgPath() { - panic(ErrWrongType) + return 0, 0, false, ErrWrongType } } diff --git a/cjson/serializer.go b/cjson/serializer.go index 2b65daa9c..45dd0845f 100644 --- a/cjson/serializer.go +++ b/cjson/serializer.go @@ -2,6 +2,7 @@ package cjson import ( "encoding/binary" + "errors" "fmt" "math" "reflect" @@ -15,6 +16,7 @@ var isLittleEndian = func() bool { var v uint16 = 0x0102 return *(*byte)(unsafe.Pointer(&v)) == 0x02 }() +var errInvalidReflectionValue = errors.New("rq: invalid reflection value") type Serializer struct { buf []byte @@ -181,10 +183,12 @@ func (s *Serializer) PutValue(v reflect.Value) error { s.PutVarCUInt(valueTuple) s.PutVarCUInt(v.Len()) for i := 0; i < v.Len(); i++ { - s.PutValue(v.Index(i)) + if err := s.PutValue(v.Index(i)); err != nil { + return err + } } default: - panic(fmt.Errorf("rq: Invalid reflection type %s", v.Kind().String())) + return errInvalidReflectionValue } return nil } diff --git a/expression.go b/expression.go index 047ae5102..3505c81b6 100644 --- a/expression.go +++ b/expression.go @@ -39,12 +39,12 @@ type SubQuery struct { SubQuery *Query } -// FlatArrayLen is the flat_array_len(field) expression; implements IFunction and IExpression. +// FlatArrayLen is the flat_array_len(field) expression. type FlatArrayLen struct { Field string } -// Now is the now(time_unit) expression; implements IFunction and IExpression. +// Now is the now(time_unit) expression. type Now struct { TimeUnit TimeUnit } @@ -87,48 +87,6 @@ func (s SubQuery) Serialize(ser *cjson.Serializer) { ser.PutVBytes(s.SubQuery.ser.Bytes()) } -func (f FlatArrayLen) FunctionType() int { - return functionFlatArrayLen -} - -func (f FlatArrayLen) Fields() []string { - return []string{f.Field} -} - -func (f FlatArrayLen) Args() []any { - return []any{} -} - -func (f Now) FunctionType() int { - return functionNow -} - -func (f Now) Fields() []string { - return []string{} -} - -func (f Now) Args() []any { - return []any{f.TimeUnit} -} - -type IFunction interface { - FunctionType() int - Fields() []string - Args() []any -} - -func SerializeFunction(fn IFunction, ser *cjson.Serializer) { - ser.PutVarCUInt(len(fn.Fields())) - for _, field := range fn.Fields() { - ser.PutVString(field) - } - ser.PutVarCUInt(len(fn.Args())) - for _, arg := range fn.Args() { - ser.PutValue(reflect.ValueOf(arg)) - } - ser.PutVarCUInt(fn.FunctionType()) -} - // Type reports expressionTypeExpression for the flat_array_len function node. func (f FlatArrayLen) Type() int { return expressionTypeExpression @@ -137,7 +95,10 @@ func (f FlatArrayLen) Type() int { // Serialize writes the function expression tag and flat_array_len payload. func (f FlatArrayLen) Serialize(ser *cjson.Serializer) { ser.PutVarCUInt(int(f.Type())) - SerializeFunction(f, ser) + ser.PutVarCUInt(1) + ser.PutVString(f.Field) + ser.PutVarCUInt(0) + ser.PutVarCUInt(functionFlatArrayLen) } // Type reports expressionTypeExpression for the now function node. @@ -148,5 +109,8 @@ func (n Now) Type() int { // Serialize writes the function expression tag and now() payload. func (n Now) Serialize(ser *cjson.Serializer) { ser.PutVarCUInt(int(n.Type())) - SerializeFunction(n, ser) + ser.PutVarCUInt(0) + ser.PutVarCUInt(1) + ser.PutVarCUInt(valueString).PutVString(string(n.TimeUnit)) + ser.PutVarCUInt(functionNow) } diff --git a/query.go b/query.go index ca2eb628c..172050250 100644 --- a/query.go +++ b/query.go @@ -3,10 +3,7 @@ package reindexer import ( "context" "errors" - "fmt" - "os" "reflect" - "runtime" "slices" "strconv" "strings" @@ -138,6 +135,28 @@ const ( defaultFetchCount = 1000 ) +var ( + errQueryNilSubQuery = errors.New("rq: nil subquery") + errQueryNilExpression = errors.New("rq: nil query expression") + errUnsupportedExpression = errors.New("rq: unsupported query expression") + errKnnEfLessThanK = errors.New("Ef should not be less than K") + errKnnNProbeLessThanOne = errors.New("Nprobe should not be less than 1") + errQueryOperationBeforeCloseBracket = errors.New("rq: operation before close bracket") + errQueryCloseBracketBeforeOpen = errors.New("rq: close bracket before open it") + errQueryNilKnnSearchParams = errors.New("rq: nil knn search params") + errQueryExecClosed = errors.New("Exec call on already closed query. You should create new Query") + errQueryExecExecuted = errors.New("Exec call on already executed query. You should create new Query") + errQueryDeleteClosed = errors.New("Delete call on already closed query. You should create new Query") + errQueryDeleteExecuted = errors.New("Delete call on already executed query. You should create new Query") + errQueryUpdateClosed = errors.New("Update call on already closed query. You should create new Query") + errQueryUpdateExecuted = errors.New("Update call on already executed query. You should create new Query") + errQueryJoinNil = errors.New("query.Join call with nil query") + errQueryJoinAlreadyJoined = errors.New("query.Join call on already joined query. You should create new Query") + errQueryMergeNil = errors.New("query.Merge call with nil query") + errQueryOnClosed = errors.New("query.On call on already closed query. You should create new Query") + errQueryOnRoot = errors.New("Can't join on root query") +) + type nsArrayEntry struct { *reindexerNamespace localCjsonState cjson.State @@ -175,12 +194,11 @@ type Query struct { jsonOffsets []int totalName string executed bool + err error fetchCount int whereEntriesCount int openedBrackets []int tx *Tx - traceNew []byte - traceClose []byte } type KnnSearchParam interface { @@ -215,14 +233,14 @@ func NewIndexBFSearchParam(baseParam BaseKnnSearchParam) (IndexBFSearchParam, er func NewIndexHnswSearchParam(ef int, baseParam BaseKnnSearchParam) (IndexHnswSearchParam, error) { if baseParam.K != nil && ef < *baseParam.K { - return IndexHnswSearchParam{}, fmt.Errorf("Ef should not be less than K") + return IndexHnswSearchParam{}, errKnnEfLessThanK } return IndexHnswSearchParam{baseParam, ef}, nil } func NewIndexIvfSearchParam(nprobe int, baseParam BaseKnnSearchParam) (IndexIvfSearchParam, error) { if nprobe < 1 { - return IndexIvfSearchParam{}, fmt.Errorf("Nprobe should not be less than 1") + return IndexIvfSearchParam{}, errKnnNProbeLessThanOne } return IndexIvfSearchParam{baseParam, nprobe}, nil } @@ -281,18 +299,6 @@ func (p IndexIvfSearchParam) serialize(ser *cjson.Serializer) { } var queryPool sync.Pool -var enableDebug bool - -func init() { - enableDebug = os.Getenv("REINDEXER_GODEBUG") != "" -} - -func mktrace(buf *[]byte) { - if *buf == nil { - *buf = make([]byte, 0x4000) - } - *buf = (*buf)[0:runtime.Stack((*buf)[0:cap((*buf))], false)] -} // Create new DB query func newQuery(db *reindexerImpl, namespace string, tx *Tx) *Query { @@ -319,14 +325,11 @@ func newQuery(db *reindexerImpl, namespace string, tx *Tx) *Query { q.closed = false q.totalName = "" q.executed = false + q.err = nil q.nsArray = q.nsArray[:0] q.whereEntriesCount = 0 q.openedBrackets = q.openedBrackets[:0] } - if enableDebug { - mktrace(&q.traceNew) - } - q.Namespace = namespace q.db = db q.nextOp = opAND @@ -352,10 +355,6 @@ func (q *Query) makeCopy(db *reindexerImpl, root *Query) *Query { if qC == nil { qC = &Query{} } - if enableDebug { - mktrace(&qC.traceNew) - } - qC.ser = cjson.NewSerializer(qC.initBuf[:0]) qC.db = db @@ -376,6 +375,7 @@ func (q *Query) makeCopy(db *reindexerImpl, root *Query) *Query { qC.jsonOffsets = append(q.jsonOffsets[:0:0], q.jsonOffsets...) qC.totalName = q.totalName qC.executed = q.executed + qC.err = q.err qC.fetchCount = q.fetchCount qC.closed = q.closed @@ -398,6 +398,18 @@ func (q *Query) makeCopy(db *reindexerImpl, root *Query) *Query { } +func (q *Query) setErr(err error) { + if err == nil { + return + } + if q.root != nil { + q = q.root + } + if q.err == nil { + q.err = err + } +} + // Where - Add where condition to DB query // 'keys' may be: // - single value @@ -412,6 +424,10 @@ func (q *Query) Where(index string, condition int, keys any) *Query { q.ser.PutVarUInt(0) return q.finishWhere() case *Query: + if v == nil { + q.setErr(errQueryNilSubQuery) + return q + } q.putSubQueryWhere(index, condition, v.ser.Bytes()) return q.finishWhere() case Query: @@ -505,7 +521,10 @@ func (q *Query) Where(index string, condition int, keys any) *Query { q.putWhereHeader(index, condition) q.ser.PutVarCUInt(len(v)) for _, value := range v { - q.ser.PutValue(reflect.ValueOf(value)) + if err := q.ser.PutValue(reflect.ValueOf(value)); err != nil { + q.setErr(err) + return q + } } return q.finishWhere() } @@ -517,13 +536,19 @@ func (q *Query) Where(index string, condition int, keys any) *Query { l := v.Len() q.ser.PutVarCUInt(l) for i := range l { - q.ser.PutValue(v.Index(i)) + if err := q.ser.PutValue(v.Index(i)); err != nil { + q.setErr(err) + return q + } } return q.finishWhere() } q.ser.PutVarCUInt(1) - q.ser.PutValue(v) + if err := q.ser.PutValue(v); err != nil { + q.setErr(err) + return q + } return q.finishWhere() } @@ -563,6 +588,10 @@ func (q *Query) finishWhere() *Query { // - slice/arrays of values // - nil for CondAny/CondEmpty func (q *Query) WhereQuery(subQuery *Query, condition int, keys any) *Query { + if subQuery == nil { + q.setErr(errQueryNilSubQuery) + return q + } q.ser.PutVarCUInt(querySubQueryCondition) q.ser.PutVarCUInt(q.nextOp) q.ser.PutVBytes(subQuery.ser.Bytes()) @@ -580,11 +609,17 @@ func (q *Query) WhereQuery(subQuery *Query, condition int, keys any) *Query { if t.Kind() == reflect.Slice || t.Kind() == reflect.Array { q.ser.PutVarCUInt(v.Len()) for i := range v.Len() { - q.ser.PutValue(v.Index(i)) + if err := q.ser.PutValue(v.Index(i)); err != nil { + q.setErr(err) + return q + } } } else { q.ser.PutVarCUInt(1) - q.ser.PutValue(v) + if err := q.ser.PutValue(v); err != nil { + q.setErr(err) + return q + } } return q } @@ -614,10 +649,12 @@ func (q *Query) OpenBracket() *Query { // CloseBracket - Close bracket for where condition to DB query func (q *Query) CloseBracket() *Query { if q.nextOp != opAND { - panic(fmt.Errorf("rq: operation before close bracket")) + q.setErr(errQueryOperationBeforeCloseBracket) + return q } if len(q.openedBrackets) < 1 { - panic(fmt.Errorf("rq: close bracket before open it")) + q.setErr(errQueryCloseBracketBeforeOpen) + return q } q.ser.PutVarCUInt(queryCloseBracket) q.openedBrackets = q.openedBrackets[:len(q.openedBrackets)-1] @@ -701,6 +738,10 @@ func (q *Query) WhereUuid(index string, condition int, keys ...string) *Query { // WhereKnn - Add where condition to DB query with float-point vector arguments. // 'index' MUST be declared as "float_vector" index in this case func (q *Query) WhereKnn(index string, vec []float32, params KnnSearchParam) *Query { + if params == nil { + q.setErr(errQueryNilKnnSearchParams) + return q + } q.ser.PutVarCUInt(queryKnnCondition).PutVString(index).PutVarCUInt(q.nextOp).PutFloatVector(vec) params.serialize(&q.ser) q.nextOp = opAND @@ -712,6 +753,10 @@ func (q *Query) WhereKnn(index string, vec []float32, params KnnSearchParam) *Qu // Before this, you need to configure "query_embedder". // 'index' MUST be declared as "float_vector" index in this case func (q *Query) WhereKnnString(index string, val string, params KnnSearchParam) *Query { + if params == nil { + q.setErr(errQueryNilKnnSearchParams) + return q + } q.ser.PutVarCUInt(queryKnnConditionExt).PutVString(index).PutVarCUInt(q.nextOp).PutVarCUInt(knnQueryDataFormatString).PutVString(val) params.serialize(&q.ser) q.nextOp = opAND @@ -774,10 +819,89 @@ func (q *Query) WhereDouble(index string, condition int, keys ...float64) *Query // Check IExpression interface and list of implementations nearby for more details. func (q *Query) WhereExpressions(left IExpression, condition int, right IExpression) *Query { q.ser.PutVarCUInt(queryExpressions) - left.Serialize(&q.ser) + + switch v := left.(type) { + case nil: + q.setErr(errQueryNilExpression) + return q + case Field: + q.ser.PutVarCUInt(expressionTypeField) + q.ser.PutVString(v.Name) + case Values: + q.ser.PutVarCUInt(expressionTypeValues) + q.ser.PutVarCUInt(len(v.Values)) + for _, value := range v.Values { + if err := q.ser.PutValue(reflect.ValueOf(value)); err != nil { + q.setErr(err) + return q + } + } + case SubQuery: + if v.SubQuery == nil { + q.setErr(errQueryNilSubQuery) + return q + } + q.ser.PutVarCUInt(expressionTypeSubQuery) + q.ser.PutVBytes(v.SubQuery.ser.Bytes()) + case FlatArrayLen: + q.ser.PutVarCUInt(expressionTypeExpression) + q.ser.PutVarCUInt(1) + q.ser.PutVString(v.Field) + q.ser.PutVarCUInt(0) + q.ser.PutVarCUInt(functionFlatArrayLen) + case Now: + q.ser.PutVarCUInt(expressionTypeExpression) + q.ser.PutVarCUInt(0) + q.ser.PutVarCUInt(1) + q.ser.PutVarCUInt(valueString).PutVString(string(v.TimeUnit)) + q.ser.PutVarCUInt(functionNow) + default: + q.setErr(errUnsupportedExpression) + return q + } + q.ser.PutVarCUInt(q.nextOp) q.ser.PutVarCUInt(condition) - right.Serialize(&q.ser) + + switch v := right.(type) { + case nil: + q.setErr(errQueryNilExpression) + return q + case Field: + q.ser.PutVarCUInt(expressionTypeField) + q.ser.PutVString(v.Name) + case Values: + q.ser.PutVarCUInt(expressionTypeValues) + q.ser.PutVarCUInt(len(v.Values)) + for _, value := range v.Values { + if err := q.ser.PutValue(reflect.ValueOf(value)); err != nil { + q.setErr(err) + return q + } + } + case SubQuery: + if v.SubQuery == nil { + q.setErr(errQueryNilSubQuery) + return q + } + q.ser.PutVarCUInt(expressionTypeSubQuery) + q.ser.PutVBytes(v.SubQuery.ser.Bytes()) + case FlatArrayLen: + q.ser.PutVarCUInt(expressionTypeExpression) + q.ser.PutVarCUInt(1) + q.ser.PutVString(v.Field) + q.ser.PutVarCUInt(0) + q.ser.PutVarCUInt(functionFlatArrayLen) + case Now: + q.ser.PutVarCUInt(expressionTypeExpression) + q.ser.PutVarCUInt(0) + q.ser.PutVarCUInt(1) + q.ser.PutVarCUInt(valueString).PutVString(string(v.TimeUnit)) + q.ser.PutVarCUInt(functionNow) + default: + q.setErr(errUnsupportedExpression) + return q + } q.whereEntriesCount++ q.nextOp = opAND @@ -865,7 +989,10 @@ func (q *Query) Sort(sortIndex string, desc bool, values ...any) *Query { q.ser.PutVarCUInt(len(values)) for i := range values { - q.ser.PutValue(reflect.ValueOf(values[i])) + if err := q.ser.PutValue(reflect.ValueOf(values[i])); err != nil { + q.setErr(err) + return q + } } return q @@ -1003,10 +1130,15 @@ func (q *Query) ExecCtx(ctx context.Context) *Iterator { q = q.root } if q.closed { - q.panicTrace("Exec call on already closed query. You should create new Query") + q.setErr(errQueryExecClosed) + return errIterator(q.err) } if q.executed { - q.panicTrace("Exec call on already executed query. You should create new Query") + q.setErr(errQueryExecExecuted) + return errIterator(q.err) + } + if q.err != nil { + return errIterator(q.err) } q.executed = true @@ -1025,10 +1157,15 @@ func (q *Query) ExecToJsonCtx(ctx context.Context, jsonRoots ...string) *JSONIte q = q.root } if q.closed { - q.panicTrace("Exec call on already closed query. You should create new Query") + q.setErr(errQueryExecClosed) + return errJSONIterator(q.err) } if q.executed { - q.panicTrace("Exec call on already executed query. You should create new Query") + q.setErr(errQueryExecExecuted) + return errJSONIterator(q.err) + } + if q.err != nil { + return errJSONIterator(q.err) } q.executed = true @@ -1046,26 +1183,17 @@ func (q *Query) close() { q = q.root } if q.closed { - q.panicTrace("Close call on already closed query") - } - if enableDebug { - mktrace(&q.traceClose) + return } for i, jq := range q.joinQueries { jq.closed = true - if enableDebug { - mktrace(&jq.traceClose) - } queryPool.Put(jq) q.joinQueries[i] = nil } for i, mq := range q.mergedQueries { mq.closed = true - if enableDebug { - mktrace(&mq.traceClose) - } queryPool.Put(mq) q.mergedQueries[i] = nil } @@ -1079,15 +1207,6 @@ func (q *Query) close() { queryPool.Put(q) } -func (q *Query) panicTrace(msg string) { - if !enableDebug { - fmt.Println("To see query allocation/close traces set REINDEXER_GODEBUG=1 environment variable!") - } else { - fmt.Printf("Query allocation trace: %s\n\nQuery close trace %s\n\n", string(q.traceNew), string(q.traceClose)) - } - panic(errors.New(msg)) -} - // Delete will execute query, and delete items, matches query // On success return number of deleted elements func (q *Query) Delete() (int, error) { @@ -1102,7 +1221,15 @@ func (q *Query) DeleteCtx(ctx context.Context) (int, error) { } if q.closed { - q.panicTrace("Delete call on already closed query. You should create new Query") + q.setErr(errQueryDeleteClosed) + return 0, q.err + } + if q.executed { + q.setErr(errQueryDeleteExecuted) + return 0, q.err + } + if q.err != nil { + return 0, q.err } defer q.close() @@ -1116,22 +1243,22 @@ func (q *Query) DeleteCtx(ctx context.Context) (int, error) { var emptyObjectJSON = []byte("{}") -func getValueJSON(value any) []byte { +func getValueJSON(value any) ([]byte, error) { if value == nil { - return emptyObjectJSON + return emptyObjectJSON, nil } if objectJSON, ok := value.([]byte); ok { - return objectJSON + return objectJSON, nil } t := reflect.TypeOf(value) if t.Kind() == reflect.Struct || t.Kind() == reflect.Map { objectJSON, err := json.Marshal(value) if err != nil { - panic(err) + return nil, err } - return objectJSON + return objectJSON, nil } - panic(errors.New("SetObject doesn't support this type of objects: " + t.Kind().String())) + return nil, errors.New("SetObject doesn't support this type of objects: " + t.Kind().String()) } // SetObject adds update of object field request for update query @@ -1158,10 +1285,20 @@ func (q *Query) SetObject(field string, values any) *Query { if isArray { jsonValues = make([][]byte, size) for i := range size { - jsonValues[i] = getValueJSON(v.Index(i).Interface()) + var err error + jsonValues[i], err = getValueJSON(v.Index(i).Interface()) + if err != nil { + q.setErr(err) + return q + } } } else if size > 0 { - jsonValue = getValueJSON(values) + var err error + jsonValue, err = getValueJSON(values) + if err != nil { + q.setErr(err) + return q + } } q.putSetObjectHeader(field, size, isArray) for i := range size { @@ -1306,6 +1443,12 @@ func (q *Query) Set(field string, values any) *Query { return q } + if values == nil { + q.putSetHeader(field, queryUpdateField, false) + q.ser.PutVarUInt(0) // size + return q + } + t := reflect.TypeOf(values) if t.Kind() == reflect.Struct || t.Kind() == reflect.Map { return q.SetObject(field, values) @@ -1325,12 +1468,7 @@ func (q *Query) Set(field string, values any) *Query { q.ser.PutVarCUInt(cmd) q.ser.PutVString(field) - if values == nil { - if cmd == queryUpdateFieldV2 { - q.ser.PutVarUInt(0) // is array - } - q.ser.PutVarUInt(0) // size - } else if t.Kind() == reflect.Slice || t.Kind() == reflect.Array { + if t.Kind() == reflect.Slice || t.Kind() == reflect.Array { if cmd == queryUpdateFieldV2 { q.ser.PutVarUInt(1) // is array } @@ -1338,7 +1476,10 @@ func (q *Query) Set(field string, values any) *Query { for i := range v.Len() { // function/value flag q.ser.PutVarUInt(0) - q.ser.PutValue(v.Index(i)) + if err := q.ser.PutValue(v.Index(i)); err != nil { + q.setErr(err) + return q + } } } else { if cmd == queryUpdateFieldV2 { @@ -1347,7 +1488,10 @@ func (q *Query) Set(field string, values any) *Query { q.ser.PutVarCUInt(1) // size // function/value flag q.ser.PutVarUInt(0) - q.ser.PutValue(v) + if err := q.ser.PutValue(v); err != nil { + q.setErr(err) + return q + } } return q } @@ -1385,7 +1529,9 @@ func (q *Query) SetExpression(field string, value string) *Query { q.ser.PutVarCUInt(1) // size q.ser.PutVarUInt(1) // is expression - q.ser.PutValue(reflect.ValueOf(value)) + if err := q.ser.PutValue(reflect.ValueOf(value)); err != nil { + q.setErr(err) + } return q } @@ -1403,7 +1549,15 @@ func (q *Query) UpdateCtx(ctx context.Context) *Iterator { q = q.root } if q.closed { - q.panicTrace("Update call on already closed query. You should create new Query") + q.setErr(errQueryUpdateClosed) + return errIterator(q.err) + } + if q.executed { + q.setErr(errQueryUpdateExecuted) + return errIterator(q.err) + } + if q.err != nil { + return errIterator(q.err) } q.executed = true @@ -1436,15 +1590,32 @@ func (q *Query) Get() (item any, found bool) { // GetCtx will execute query and return first item. Generates panic on error func (q *Query) GetCtx(ctx context.Context) (item any, found bool) { + item, found, err := q.GetErrCtx(ctx) + if err != nil { + panic(err) + } + return item, found +} + +// GetErr will execute query and return first item or error. +func (q *Query) GetErr() (item any, found bool, err error) { + return q.GetErrCtx(context.Background()) +} + +// GetErrCtx will execute query and return first item or error. +func (q *Query) GetErrCtx(ctx context.Context) (item any, found bool, err error) { if q.root != nil { q = q.root } - iter := q.Limit(1).MustExecCtx(ctx) + iter := q.Limit(1).ExecCtx(ctx) defer iter.Close() + if err := iter.Error(); err != nil { + return nil, false, err + } if iter.Next() { - return iter.Object(), true + return iter.Object(), true, nil } - return nil, false + return nil, false, nil } // GetJson will execute query, and return 1 st item. Generates panic on error @@ -1454,19 +1625,33 @@ func (q *Query) GetJson() (json []byte, found bool) { // GetJsonCtx will execute query, and return first item. Generates panic on error func (q *Query) GetJsonCtx(ctx context.Context) (json []byte, found bool) { + json, found, err := q.GetJsonErrCtx(ctx) + if err != nil { + panic(err) + } + return json, found +} + +// GetJsonErr will execute query, and return first item as json or error. +func (q *Query) GetJsonErr() (json []byte, found bool, err error) { + return q.GetJsonErrCtx(context.Background()) +} + +// GetJsonErrCtx will execute query, and return first item as json or error. +func (q *Query) GetJsonErrCtx(ctx context.Context) (json []byte, found bool, err error) { if q.root != nil { q = q.root } it := q.Limit(1).ExecToJsonCtx(ctx) defer it.Close() if it.Error() != nil { - panic(it.Error()) + return nil, false, it.Error() } if !it.Next() { - return nil, false + return nil, false, nil } - return it.JSON(), true + return it.JSON(), true, nil } // Join joins 2 queries @@ -1474,8 +1659,13 @@ func (q *Query) join(q2 *Query, field string, joinType int) *Query { if q.root != nil { q = q.root } + if q2 == nil { + q.setErr(errQueryJoinNil) + return q + } if q2.root != nil { - panic(errors.New("query.Join call on already joined query. You should create new Query")) + q.setErr(errQueryJoinAlreadyJoined) + return q } if joinType != leftJoin { q.ser.PutVarCUInt(queryJoinCondition) @@ -1551,6 +1741,10 @@ func (q *Query) Merge(q2 *Query) *Query { if q.root != nil { q = q.root } + if q2 == nil { + q.setErr(errQueryMergeNil) + return q + } if q2.root != nil { q2 = q2.root } @@ -1566,10 +1760,12 @@ func (q *Query) Merge(q2 *Query) *Query { // `joinIndex` parameter specifies which field from namespace for the latest join query issued on `q` should be used during join func (q *Query) On(index string, condition int, joinIndex string) *Query { if q.closed { - q.panicTrace("query.On call on already closed query. You should create new Query") + q.setErr(errQueryOnClosed) + return q } if q.root == nil { - panic(fmt.Errorf("Can't join on root query")) + q.setErr(errQueryOnRoot) + return q } q.ser.PutVarCUInt(queryJoinOn) q.ser.PutVarCUInt(q.nextOp) diff --git a/reflect.go b/reflect.go index ad48cc85c..81f6d6203 100644 --- a/reflect.go +++ b/reflect.go @@ -1,6 +1,7 @@ package reindexer import ( + "errors" "fmt" "maps" "reflect" @@ -30,6 +31,11 @@ var collateModes = map[string]int{ "collate_custom": CollateCustom, } +var ( + errDuplicateCollateMode = errors.New("collate mode is already set") + errInvalidExpireAfter = errors.New("'ExpireAfter' should be an integer value") +) + type indexOptions struct { isArray bool isAppenable bool @@ -341,7 +347,11 @@ func parseIndexesImpl(indexDefs *[]bindings.IndexDef, st reflect.Type, subArray strings.SplitN(field.Tag.Get("reindex"), ",", 3), field.Name) } - indexDef := makeIndexDef(parseCompositeName(reindexPath), parseCompositeJsonPaths(reindexPath), idxType, "composite", opts, CollateNone, "", parseExpireAfter(expireAfter), nil) + expireAfterValue, err := parseExpireAfter(expireAfter) + if err != nil { + return err + } + indexDef := makeIndexDef(parseCompositeName(reindexPath), parseCompositeJsonPaths(reindexPath), idxType, "composite", opts, CollateNone, "", expireAfterValue, nil) if err := indexDefAppend(indexDefs, indexDef, opts.isAppenable); err != nil { return err } @@ -362,7 +372,10 @@ func parseIndexesImpl(indexDefs *[]bindings.IndexDef, st reflect.Type, subArray return err } } else if len(idxName) > 0 { - collateMode, sortOrderLetters := parseCollate(&idxSettings) + collateMode, sortOrderLetters, err := parseCollate(&idxSettings) + if err != nil { + return err + } var fieldType string if idxType == "rtree" { fieldType = "point" @@ -381,7 +394,11 @@ func parseIndexesImpl(indexDefs *[]bindings.IndexDef, st reflect.Type, subArray reindexPath, field.Name, jsonPath) } - indexDef := makeIndexDef(reindexPath, []string{jsonPath}, idxType, fieldType, opts, collateMode, sortOrderLetters, parseExpireAfter(expireAfter), nil) + expireAfterValue, err := parseExpireAfter(expireAfter) + if err != nil { + return err + } + indexDef := makeIndexDef(reindexPath, []string{jsonPath}, idxType, fieldType, opts, collateMode, sortOrderLetters, expireAfterValue, nil) if err := indexDefAppend(indexDefs, indexDef, opts.isAppenable); err != nil { return err } @@ -445,7 +462,7 @@ func parseCompositeJsonPaths(indexName string) []string { return strings.Split(indexConents[0], "+") } -func parseCollate(idxSettingsBuf *[]string) (int, string) { +func parseCollate(idxSettingsBuf *[]string) (int, string, error) { newIdxSettingsBuf := make([]string, 0) collateMode := CollateNone @@ -457,7 +474,7 @@ func parseCollate(idxSettingsBuf *[]string) (int, string) { if newCollateMode, ok := collateModes[kvIdxSettings[0]]; ok { if collateMode != CollateNone { - panic(fmt.Errorf("collate mode is already set to '%d'. Misunderstanding '%s'", collateMode, idxSetting)) + return 0, "", errDuplicateCollateMode } collateMode = newCollateMode @@ -474,19 +491,19 @@ func parseCollate(idxSettingsBuf *[]string) (int, string) { *idxSettingsBuf = newIdxSettingsBuf - return collateMode, sortOrderLetters + return collateMode, sortOrderLetters, nil } -func parseExpireAfter(str string) int { +func parseExpireAfter(str string) (int, error) { expireAfter := 0 if len(str) > 0 { var err error expireAfter, err = strconv.Atoi(str) if err != nil { - panic(fmt.Errorf("'ExpireAfter' should be an integer value")) + return 0, errInvalidExpireAfter } } - return expireAfter + return expireAfter, nil } func parseByKeyWord(idxSettingsBuf *[]string, keyWord string) bool { diff --git a/test/namespace_parse_error_test.go b/test/namespace_parse_error_test.go new file mode 100644 index 000000000..647ddb4b7 --- /dev/null +++ b/test/namespace_parse_error_test.go @@ -0,0 +1,34 @@ +package reindexer + +import ( + "testing" + + rx "github.com/restream/reindexer/v5" + "github.com/stretchr/testify/require" +) + +type badCollateNamespaceItem struct { + ID int `json:"id" reindex:"id,,pk"` + Name string `json:"name" reindex:"name,tree,collate_ascii,collate_utf8"` +} + +type badExpireAfterNamespaceItem struct { + ID int `json:"id" reindex:"id,,pk"` + Date int64 `json:"date" reindex:"date,ttl,,expire_after=abc"` +} + +func TestOpenNamespaceInvalidCollateReturnsError(t *testing.T) { + var err error + require.NotPanics(t, func() { + err = DB.Reindexer.OpenNamespace("test_bad_collate_namespace", rx.DefaultNamespaceOptions(), badCollateNamespaceItem{}) + }) + require.Error(t, err) +} + +func TestOpenNamespaceInvalidExpireAfterReturnsError(t *testing.T) { + var err error + require.NotPanics(t, func() { + err = DB.Reindexer.OpenNamespace("test_bad_expire_after_namespace", rx.DefaultNamespaceOptions(), badExpireAfterNamespaceItem{}) + }) + require.Error(t, err) +} diff --git a/test/pack_item_test.go b/test/pack_item_test.go index af8f5a5d0..336276ba1 100644 --- a/test/pack_item_test.go +++ b/test/pack_item_test.go @@ -2,7 +2,9 @@ package reindexer import ( "testing" + "time" + rx "github.com/restream/reindexer/v5" "github.com/stretchr/testify/require" ) @@ -26,7 +28,34 @@ func TestPackItemTypeCheck(t *testing.T) { require.NoError(t, DB.Upsert(ns, packItemTypeCheckItem{ID: 1})) require.NoError(t, DB.Upsert(ns, &packItemTypeCheckItem{ID: 2})) require.NoError(t, DB.Upsert(ns, []byte(`{"id":3}`))) - require.Panics(t, func() { - _ = DB.Upsert(ns, packItemWrongNamedItem{ID: 4}) + require.Equal(t, rx.ErrWrongType, DB.Upsert(ns, packItemWrongNamedItem{ID: 4})) + require.NotPanics(t, func() { + require.Equal(t, rx.ErrWrongType, DB.Upsert(ns, packItemWrongNamedItem{ID: 5})) }) } + +func TestPackItemTypeCheckTx(t *testing.T) { + const ns = testPackItemTypeCheckNs + + tx := DB.Reindexer.MustBeginTx(ns) + defer tx.Rollback() + require.Equal(t, rx.ErrWrongType, tx.Upsert(packItemWrongNamedItem{ID: 1})) +} + +func TestPackItemTypeCheckTxAsync(t *testing.T) { + const ns = testPackItemTypeCheckNs + + tx := DB.Reindexer.MustBeginTx(ns) + defer tx.Rollback() + errCh := make(chan error, 1) + require.Equal(t, rx.ErrWrongType, tx.UpsertAsync(packItemWrongNamedItem{ID: 1}, func(err error) { + errCh <- err + })) + tx.AwaitResults() + select { + case err := <-errCh: + require.Equal(t, rx.ErrWrongType, err) + case <-time.After(time.Second): + t.Fatal("timeout waiting for async upsert completion") + } +} diff --git a/test/query_error_test.go b/test/query_error_test.go new file mode 100644 index 000000000..62eb58c01 --- /dev/null +++ b/test/query_error_test.go @@ -0,0 +1,112 @@ +package reindexer + +import ( + "testing" + + rx "github.com/restream/reindexer/v5" + "github.com/restream/reindexer/v5/cjson" + "github.com/stretchr/testify/require" +) + +const testQueryErrorNs = "test_query_error_namespace" + +type queryErrorItem struct { + ID int `json:"id" reindex:"id,,pk"` + Name string `json:"name" reindex:"name"` +} + +type unsupportedQueryExpression struct{} + +func (unsupportedQueryExpression) Type() int { + return 0 +} + +func (unsupportedQueryExpression) Serialize(*cjson.Serializer) { + panic("unsupported expression Serialize should not be called") +} + +func init() { + tnamespaces[testQueryErrorNs] = queryErrorItem{} +} + +func TestQueryGetErr(t *testing.T) { + const ns = testQueryErrorNs + + require.NoError(t, DB.Upsert(ns, queryErrorItem{ID: 1, Name: "one"})) + + item, found, err := DB.Reindexer.Query(ns).WhereInt("id", rx.EQ, 1).GetErr() + require.NoError(t, err) + require.True(t, found) + require.Equal(t, 1, item.(*queryErrorItem).ID) + + item, found, err = DB.Reindexer.Query(ns).WhereInt("id", rx.EQ, 2).GetErr() + require.NoError(t, err) + require.False(t, found) + require.Nil(t, item) + + json, found, err := DB.Reindexer.Query(ns).WhereInt("id", rx.EQ, 1).GetJsonErr() + require.NoError(t, err) + require.True(t, found) + require.Contains(t, string(json), `"id":1`) + + item, found, err = DB.Reindexer.Query(ns).Sort("id", false, struct{}{}).GetErr() + require.Error(t, err) + require.False(t, found) + require.Nil(t, item) + + json, found, err = DB.Reindexer.Query(ns).Sort("id", false, struct{}{}).GetJsonErr() + require.Error(t, err) + require.False(t, found) + require.Nil(t, json) +} + +func TestQueryGetWrappersStillPanicOnError(t *testing.T) { + const ns = testQueryErrorNs + + require.Panics(t, func() { + DB.Reindexer.Query(ns).Sort("id", false, struct{}{}).Get() + }) + require.Panics(t, func() { + DB.Reindexer.Query(ns).Sort("id", false, struct{}{}).GetJson() + }) +} + +func TestQueryBuilderErrors(t *testing.T) { + const ns = testQueryErrorNs + + it := DB.Reindexer.Query(ns).CloseBracket().Exec() + require.Error(t, it.Error()) + it.Close() + + it = DB.Reindexer.Query(ns).Sort("id", false, struct{}{}).Exec() + require.Error(t, it.Error()) + it.Close() + + it = DB.Reindexer.Query(ns).Sort("id", false, nil).Exec() + require.Error(t, it.Error()) + it.Close() + + it = DB.Reindexer.Query(ns).On("id", rx.EQ, "id").Exec() + require.Error(t, it.Error()) + it.Close() + + require.NotPanics(t, func() { + it = DB.Reindexer.Query(ns). + WhereExpressions(unsupportedQueryExpression{}, rx.EQ, rx.Values{Values: []any{1}}). + Exec() + }) + require.Error(t, it.Error()) + it.Close() +} + +func TestQueryRepeatedExecReturnsError(t *testing.T) { + const ns = testQueryErrorNs + + q := DB.Reindexer.Query(ns).WhereInt("id", rx.EQ, 1) + it := q.Exec() + require.NoError(t, it.Error()) + it2 := q.Exec() + require.Error(t, it2.Error()) + it2.Close() + it.Close() +}