dapr
/
components-contrib
mirror of https://github.com/dapr/components-contrib.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add new filters type in the query of the state component (#3218) 

Signed-off-by: Luigi Rende <luigirende@gmail.com>
Co-authored-by: Bernd Verst <github@bernd.dev>
This commit is contained in:
luigirende 2023-12-06 21:51:27 +01:00 committed by GitHub
parent 87aea87e95
commit a7c64f4e8d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 885 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
common/component/postgresql/postgresql_query.go
View File

@ -40,6 +40,46 @@ func (q *Query) VisitEQ(f *query.EQ) (string, error) {
return q.whereFieldEqual(f.Key, f.Val), nil
}
func (q *Query) VisitNEQ(f *query.NEQ) (string, error) {
return q.whereFieldNotEqual(f.Key, f.Val), nil
}
func (q *Query) VisitGT(f *query.GT) (string, error) {
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", v)
default:
return q.whereFieldGreaterThan(f.Key, v), nil
}
}
func (q *Query) VisitGTE(f *query.GTE) (string, error) {
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", v)
default:
return q.whereFieldGreaterThanEqual(f.Key, v), nil
}
}
func (q *Query) VisitLT(f *query.LT) (string, error) {
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", v)
default:
return q.whereFieldLessThan(f.Key, v), nil
}
}
func (q *Query) VisitLTE(f *query.LTE) (string, error) {
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", v)
default:
return q.whereFieldLessThanEqual(f.Key, v), nil
}
}
func (q *Query) VisitIN(f *query.IN) (string, error) {
if len(f.Vals) == 0 {
return "", fmt.Errorf("empty IN operator for key %q", f.Key)
@ -70,6 +110,31 @@ func (q *Query) visitFilters(op string, filters []query.Filter) (string, error)
return "", err
}
arr = append(arr, str)
case *query.NEQ:
if str, err = q.VisitNEQ(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.GT:
if str, err = q.VisitGT(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.GTE:
if str, err = q.VisitGTE(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.LT:
if str, err = q.VisitLT(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.LTE:
if str, err = q.VisitLTE(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.IN:
if str, err = q.VisitIN(f); err != nil {
return "", err
@ -214,3 +279,38 @@ func (q *Query) whereFieldEqual(key string, value interface{}) string {
query := filterField + "=$" + strconv.Itoa(position)
return query
}
func (q *Query) whereFieldNotEqual(key string, value interface{}) string {
position := q.addParamValueAndReturnPosition(value)
filterField := translateFieldToFilter(key)
query := filterField + "!=$" + strconv.Itoa(position)
return query
}
func (q *Query) whereFieldGreaterThan(key string, value interface{}) string {
position := q.addParamValueAndReturnPosition(value)
filterField := translateFieldToFilter(key)
query := filterField + ">$" + strconv.Itoa(position)
return query
}
func (q *Query) whereFieldGreaterThanEqual(key string, value interface{}) string {
position := q.addParamValueAndReturnPosition(value)
filterField := translateFieldToFilter(key)
query := filterField + ">=$" + strconv.Itoa(position)
return query
}
func (q *Query) whereFieldLessThan(key string, value interface{}) string {
position := q.addParamValueAndReturnPosition(value)
filterField := translateFieldToFilter(key)
query := filterField + "<$" + strconv.Itoa(position)
return query
}
func (q *Query) whereFieldLessThanEqual(key string, value interface{}) string {
position := q.addParamValueAndReturnPosition(value)
filterField := translateFieldToFilter(key)
query := filterField + "<=$" + strconv.Itoa(position)
return query
}

8
common/component/postgresql/postgresql_query_test.go
View File

@ -49,10 +49,18 @@ func TestPostgresqlQueryBuildQuery(t *testing.T) {
input: "../../../tests/state/query/q4.json",
query: "SELECT key, value, xmin as etag FROM state WHERE (value->'person'->>'org'=$1 OR (value->'person'->>'org'=$2 AND (value->>'state'=$3 OR value->>'state'=$4))) ORDER BY value->>'state' DESC, value->'person'->>'name' LIMIT 2",
},
{
input: "../../../tests/state/query/q4-notequal.json",
query: "SELECT key, value, xmin as etag FROM state WHERE (value->'person'->>'org'=$1 OR (value->'person'->>'org'!=$2 AND (value->>'state'=$3 OR value->>'state'=$4))) ORDER BY value->>'state' DESC, value->'person'->>'name' LIMIT 2",
},
{
input: "../../../tests/state/query/q5.json",
query: "SELECT key, value, xmin as etag FROM state WHERE (value->'person'->>'org'=$1 AND (value->'person'->>'name'=$2 OR (value->>'state'=$3 OR value->>'state'=$4))) ORDER BY value->>'state' DESC, value->'person'->>'name' LIMIT 2",
},
{
input: "../../../tests/state/query/q8.json",
query: "SELECT key, value, xmin as etag FROM state WHERE (value->'person'->>'org'>=$1 OR (value->'person'->>'org'<$2 AND (value->>'state'=$3 OR value->>'state'=$4))) ORDER BY value->>'state' DESC, value->'person'->>'name' LIMIT 2",
},
}
for _, test := range tests {
data, err := os.ReadFile(test.input)

37
state/README.md
View File

@ -80,6 +80,31 @@ type FilterEQ struct {
Val interface{}
}
type FilterNEQ struct {
Key string
Val interface{}
}
type FilterGT struct {
Key string
Val interface{}
}
type FilterGTE struct {
Key string
Val interface{}
}
type FilterLT struct {
Key string
Val interface{}
}
type FilterLTE struct {
Key string
Val interface{}
}
type FilterIN struct {
Key string
Vals []interface{}
@ -100,6 +125,16 @@ To simplify the process of query translation, we leveraged [visitor design patte
type Visitor interface {
// returns "equal" expression
VisitEQ(*FilterEQ) (string, error)
// returns "not equal" expression
VisitNEQ(*FilterNEQ) (string, error)
// returns "greater than" expression
VisitGT(*FilterGT) (string, error)
// returns "greater than equal" expression
VisitGTE(*FilterGTE) (string, error)
// returns "less than" expression
VisitLT(*FilterLT) (string, error)
// returns "less than equal" expression
VisitLTE(*FilterLTE) (string, error)
// returns "in" expression
VisitIN(*FilterIN) (string, error)
// returns "and" expression
@ -152,4 +187,4 @@ func (m *MyComponent) Query(req *state.QueryRequest) (*state.QueryResponse, erro
}
```
Some of the examples of State Query API implementation are [MongoDB](./mongodb/mongodb_query.go) and [CosmosDB](./azure/cosmosdb/cosmosdb_query.go) state store components.
Some of the examples of State Query API implementation are [Redis](./redis/redis_query.go), [MongoDB](./mongodb/mongodb_query.go) and [CosmosDB](./azure/cosmosdb/cosmosdb_query.go) state store components.

109
state/azure/cosmosdb/cosmosdb_query.go
View File

@ -50,6 +50,76 @@ func (q *Query) VisitEQ(f *query.EQ) (string, error) {
return replaceKeywords("c.value."+f.Key) + " = " + name, nil
}
func (q *Query) VisitNEQ(f *query.NEQ) (string, error) {
// <key> != <val>
val, ok := f.Val.(string)
if !ok {
return "", fmt.Errorf("unsupported type of value %#v; expected string", f.Val)
}
name := q.setNextParameter(val)
return replaceKeywords("c.value."+f.Key) + " != " + name, nil
}
func (q *Query) VisitGT(f *query.GT) (string, error) {
// <key> > <val>
var name string
switch value := f.Val.(type) {
case int:
name = q.setNextParameterInt(value)
case float64:
name = q.setNextParameterFloat(value)
default:
return "", fmt.Errorf("unsupported type of value %#v; expected number", f.Val)
}
return replaceKeywords("c.value."+f.Key) + " > " + name, nil
}
func (q *Query) VisitGTE(f *query.GTE) (string, error) {
// <key> >= <val>
var name string
switch value := f.Val.(type) {
case int:
name = q.setNextParameterInt(value)
case float64:
name = q.setNextParameterFloat(value)
default:
return "", fmt.Errorf("unsupported type of value %#v; expected number", f.Val)
}
return replaceKeywords("c.value."+f.Key) + " >= " + name, nil
}
func (q *Query) VisitLT(f *query.LT) (string, error) {
// <key> < <val>
var name string
switch value := f.Val.(type) {
case int:
name = q.setNextParameterInt(value)
case float64:
name = q.setNextParameterFloat(value)
default:
return "", fmt.Errorf("unsupported type of value %#v; expected number", f.Val)
}
return replaceKeywords("c.value."+f.Key) + " < " + name, nil
}
func (q *Query) VisitLTE(f *query.LTE) (string, error) {
// <key> <= <val>
var name string
switch value := f.Val.(type) {
case int:
name = q.setNextParameterInt(value)
case float64:
name = q.setNextParameterFloat(value)
default:
return "", fmt.Errorf("unsupported type of value %#v; expected number", f.Val)
}
return replaceKeywords("c.value."+f.Key) + " <= " + name, nil
}
func (q *Query) VisitIN(f *query.IN) (string, error) {
// <key> IN ( <val1>, <val2>, ... , <valN> )
if len(f.Vals) == 0 {
@ -80,6 +150,31 @@ func (q *Query) visitFilters(op string, filters []query.Filter) (string, error)
return "", err
}
arr = append(arr, str)
case *query.NEQ:
if str, err = q.VisitNEQ(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.GT:
if str, err = q.VisitGT(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.GTE:
if str, err = q.VisitGTE(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.LT:
if str, err = q.VisitLT(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.LTE:
if str, err = q.VisitLTE(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.IN:
if str, err = q.VisitIN(f); err != nil {
return "", err
@ -144,6 +239,20 @@ func (q *Query) setNextParameter(val string) string {
return pname
}
func (q *Query) setNextParameterInt(val int) string {
pname := fmt.Sprintf("@__param__%d__", len(q.query.parameters))
q.query.parameters = append(q.query.parameters, azcosmos.QueryParameter{Name: pname, Value: val})
return pname
}
func (q *Query) setNextParameterFloat(val float64) string {
pname := fmt.Sprintf("@__param__%d__", len(q.query.parameters))
q.query.parameters = append(q.query.parameters, azcosmos.QueryParameter{Name: pname, Value: val})
return pname
}
func (q *Query) execute(ctx context.Context, client *azcosmos.ContainerClient) ([]state.QueryItem, string, error) {
opts := &azcosmos.QueryOptions{}

48
state/azure/cosmosdb/cosmosdb_query_test.go
View File

@ -126,6 +126,54 @@ func TestCosmosDbQuery(t *testing.T) {
},
},
},
{
input: "../../../tests/state/query/q4-notequal.json",
query: InternalQuery{
query: "SELECT * FROM c WHERE c['value']['person']['org'] = @__param__0__ OR (c['value']['person']['org'] != @__param__1__ AND c['value']['state'] IN (@__param__2__, @__param__3__)) ORDER BY c['value']['state'] DESC, c['value']['person']['name'] ASC",
parameters: []azcosmos.QueryParameter{
{
Name: "@__param__0__",
Value: "A",
},
{
Name: "@__param__1__",
Value: "B",
},
{
Name: "@__param__2__",
Value: "CA",
},
{
Name: "@__param__3__",
Value: "WA",
},
},
},
},
{
input: "../../../tests/state/query/q8.json",
query: InternalQuery{
query: "SELECT * FROM c WHERE c['value']['person']['org'] >= @__param__0__ OR (c['value']['person']['org'] < @__param__1__ AND c['value']['state'] IN (@__param__2__, @__param__3__)) ORDER BY c['value']['state'] DESC, c['value']['person']['name'] ASC",
parameters: []azcosmos.QueryParameter{
{
Name: "@__param__0__",
Value: 123.0,
},
{
Name: "@__param__1__",
Value: 10.0,
},
{
Name: "@__param__2__",
Value: "CA",
},
{
Name: "@__param__3__",
Value: "WA",
},
},
},
},
}
for _, test := range tests {
data, err := os.ReadFile(test.input)

75
state/mongodb/mongodb_query.go
View File

@ -45,6 +45,56 @@ func (q *Query) VisitEQ(f *query.EQ) (string, error) {
}
}
func (q *Query) VisitNEQ(f *query.NEQ) (string, error) {
// { <key>: <val> }
switch v := f.Val.(type) {
case string:
return fmt.Sprintf(`{ "value.%s": {"$ne": %q} }`, f.Key, v), nil
default:
return fmt.Sprintf(`{ "value.%s": {"$ne": %v} }`, f.Key, v), nil
}
}
func (q *Query) VisitGT(f *query.GT) (string, error) {
// { <key>: <val> }
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", f.Val)
default:
return fmt.Sprintf(`{ "value.%s": {"$gt": %v} }`, f.Key, v), nil
}
}
func (q *Query) VisitGTE(f *query.GTE) (string, error) {
// { <key>: <val> }
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", f.Val)
default:
return fmt.Sprintf(`{ "value.%s": {"$gte": %v} }`, f.Key, v), nil
}
}
func (q *Query) VisitLT(f *query.LT) (string, error) {
// { <key>: <val> }
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", f.Val)
default:
return fmt.Sprintf(`{ "value.%s": {"$lt": %v} }`, f.Key, v), nil
}
}
func (q *Query) VisitLTE(f *query.LTE) (string, error) {
// { <key>: <val> }
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", f.Val)
default:
return fmt.Sprintf(`{ "value.%s": {"$lte": %v} }`, f.Key, v), nil
}
}
func (q *Query) VisitIN(f *query.IN) (string, error) {
// { $in: [ <val1>, <val2>, ... , <valN> ] }
if len(f.Vals) == 0 {
@ -81,6 +131,31 @@ func (q *Query) visitFilters(op string, filters []query.Filter) (string, error)
return "", err
}
arr = append(arr, str)
case *query.NEQ:
if str, err = q.VisitNEQ(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.GT:
if str, err = q.VisitGT(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.GTE:
if str, err = q.VisitGTE(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.LT:
if str, err = q.VisitLT(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.LTE:
if str, err = q.VisitLTE(f); err != nil {
return "", err
}
arr = append(arr, str)
case *query.IN:
if str, err = q.VisitIN(f); err != nil {
return "", err

8
state/mongodb/mongodb_query_test.go
View File

@ -49,6 +49,14 @@ func TestMongoQuery(t *testing.T) {
input: "../../tests/state/query/q6.json",
query: `{ "$or": [ { "value.person.id": 123 }, { "$and": [ { "value.person.org": "B" }, { "value.person.id": { "$in": [ 567, 890 ] } } ] } ] }`,
},
{
input: "../../tests/state/query/q6-notequal.json",
query: `{ "$or": [ { "value.person.id": 123 }, { "$and": [ { "value.person.org": {"$ne": "B"} }, { "value.person.id": { "$in": [ 567, 890 ] } } ] } ] }`,
},
{
input: "../../tests/state/query/q7.json",
query: `{ "$or": [ { "value.person.id": {"$lt": 123} }, { "$and": [ { "value.person.org": {"$gte": 2} }, { "value.person.id": { "$in": [ 567, 890 ] } } ] } ] }`,
},
}
for _, test := range tests {
data, err := os.ReadFile(test.input)

130
state/query/filter.go
View File

@ -36,6 +36,31 @@ func ParseFilter(obj interface{}) (Filter, error) {
f := &EQ{}
err := f.Parse(v)
return f, err
case "NEQ":
f := &NEQ{}
err := f.Parse(v)
return f, err
case "GT":
f := &GT{}
err := f.Parse(v)
return f, err
case "GTE":
f := &GTE{}
err := f.Parse(v)
return f, err
case "LT":
f := &LT{}
err := f.Parse(v)
return f, err
case "LTE":
f := &LTE{}
err := f.Parse(v)
return f, err
case "IN":
f := &IN{}
@ -81,6 +106,111 @@ func (f *EQ) Parse(obj interface{}) error {
return nil
}
type NEQ struct {
Key string
Val interface{}
}
func (f *NEQ) Parse(obj interface{}) error {
m, ok := obj.(map[string]interface{})
if !ok {
return fmt.Errorf("NEQ filter must be a map")
}
if len(m) != 1 {
return fmt.Errorf("NEQ filter must contain a single key/value pair")
}
for k, v := range m {
f.Key = k
f.Val = v
}
return nil
}
type GT struct {
Key string
Val interface{}
}
func (f *GT) Parse(obj interface{}) error {
m, ok := obj.(map[string]interface{})
if !ok {
return fmt.Errorf("GT filter must be a map")
}
if len(m) != 1 {
return fmt.Errorf("GT filter must contain a single key/value pair")
}
for k, v := range m {
f.Key = k
f.Val = v
}
return nil
}
type GTE struct {
Key string
Val interface{}
}
func (f *GTE) Parse(obj interface{}) error {
m, ok := obj.(map[string]interface{})
if !ok {
return fmt.Errorf("GTE filter must be a map")
}
if len(m) != 1 {
return fmt.Errorf("GTE filter must contain a single key/value pair")
}
for k, v := range m {
f.Key = k
f.Val = v
}
return nil
}
type LT struct {
Key string
Val interface{}
}
func (f *LT) Parse(obj interface{}) error {
m, ok := obj.(map[string]interface{})
if !ok {
return fmt.Errorf("LT filter must be a map")
}
if len(m) != 1 {
return fmt.Errorf("LT filter must contain a single key/value pair")
}
for k, v := range m {
f.Key = k
f.Val = v
}
return nil
}
type LTE struct {
Key string
Val interface{}
}
func (f *LTE) Parse(obj interface{}) error {
m, ok := obj.(map[string]interface{})
if !ok {
return fmt.Errorf("LTE filter must be a map")
}
if len(m) != 1 {
return fmt.Errorf("LTE filter must contain a single key/value pair")
}
for k, v := range m {
f.Key = k
f.Val = v
}
return nil
}
type IN struct {
Key string
Vals []interface{}

20
state/query/query.go
View File

@ -53,6 +53,16 @@ type Query struct {
type Visitor interface {
// returns "equal" expression
VisitEQ(*EQ) (string, error)
// returns "not equal" expression
VisitNEQ(*NEQ) (string, error)
// returns "greater than" expression
VisitGT(*GT) (string, error)
// returns "greater than equal" expression
VisitGTE(*GTE) (string, error)
// returns "less than" expression
VisitLT(*LT) (string, error)
// returns "less than equal" expression
VisitLTE(*LTE) (string, error)
// returns "in" expression
VisitIN(*IN) (string, error)
// returns "and" expression
@ -89,6 +99,16 @@ func (h *Builder) buildFilter(filter Filter) (string, error) {
switch f := filter.(type) {
case *EQ:
return h.visitor.VisitEQ(f)
case *NEQ:
return h.visitor.VisitNEQ(f)
case *GT:
return h.visitor.VisitGT(f)
case *GTE:
return h.visitor.VisitGTE(f)
case *LT:
return h.visitor.VisitLT(f)
case *LTE:
return h.visitor.VisitLTE(f)
case *IN:
return h.visitor.VisitIN(f)
case *OR:

100
state/redis/redis_query.go
View File

@ -66,6 +66,81 @@ func (q *Query) VisitEQ(f *query.EQ) (string, error) {
}
}
func (q *Query) VisitNEQ(f *query.NEQ) (string, error) {
// string: @<key>:(<val>)
// numeric: @<key>:[<val> <val>]
alias, err := q.getAlias(f.Key)
if err != nil {
return "", err
}
switch v := f.Val.(type) {
case string:
return fmt.Sprintf("@%s:(%s)", alias, v), nil
default:
return fmt.Sprintf("@%s:[%v %v]", alias, v, v), nil
}
}
func (q *Query) VisitGT(f *query.GT) (string, error) {
// numeric: @<key>:[(<val> +inf]
alias, err := q.getAlias(f.Key)
if err != nil {
return "", err
}
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", f.Val)
default:
return fmt.Sprintf("@%s:[(%v +inf]", alias, v), nil
}
}
func (q *Query) VisitGTE(f *query.GTE) (string, error) {
// numeric: @<key>:[<val> +inf]
alias, err := q.getAlias(f.Key)
if err != nil {
return "", err
}
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", f.Val)
default:
return fmt.Sprintf("@%s:[%v +inf]", alias, v), nil
}
}
func (q *Query) VisitLT(f *query.LT) (string, error) {
// numeric: @<key>:[-inf <val>)]
alias, err := q.getAlias(f.Key)
if err != nil {
return "", err
}
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", f.Val)
default:
return fmt.Sprintf("@%s:[-inf (%v]", alias, v), nil
}
}
func (q *Query) VisitLTE(f *query.LTE) (string, error) {
// numeric: @<key>:[-inf <val>]
alias, err := q.getAlias(f.Key)
if err != nil {
return "", err
}
switch v := f.Val.(type) {
case string:
return "", fmt.Errorf("unsupported type of value %s; string type not permitted", f.Val)
default:
return fmt.Sprintf("@%s:[-inf %v]", alias, v), nil
}
}
func (q *Query) VisitIN(f *query.IN) (string, error) {
// string: @<key>:(<val1>|<val2>...)
// numeric: replace with OR
@ -116,6 +191,31 @@ func (q *Query) visitFilters(op string, filters []query.Filter) (string, error)
return "", err
}
arr = append(arr, fmt.Sprintf("(%s)", str))
case *query.NEQ:
if str, err = q.VisitNEQ(f); err != nil {
return "", err
}
arr = append(arr, fmt.Sprintf("-(%s)", str))
case *query.GT:
if str, err = q.VisitGT(f); err != nil {
return "", err
}
arr = append(arr, fmt.Sprintf("(%s)", str))
case *query.GTE:
if str, err = q.VisitGTE(f); err != nil {
return "", err
}
arr = append(arr, fmt.Sprintf("(%s)", str))
case *query.LT:
if str, err = q.VisitLT(f); err != nil {
return "", err
}
arr = append(arr, fmt.Sprintf("(%s)", str))
case *query.LTE:
if str, err = q.VisitLTE(f); err != nil {
return "", err
}
arr = append(arr, fmt.Sprintf("(%s)", str))
case *query.IN:
if str, err = q.VisitIN(f); err != nil {
return "", err

8
state/redis/redis_query_test.go
View File

@ -46,6 +46,14 @@ func TestMongoQuery(t *testing.T) {
input: "../../tests/state/query/q6.json",
query: []interface{}{"((@id:[123 123])|((@org:(B)) (((@id:[567 567])|(@id:[890 890])))))", "SORTBY", "id", "LIMIT", "0", "2"},
},
{
input: "../../tests/state/query/q6-notequal.json",
query: []interface{}{"((@id:[123 123])|(-(@org:(B)) (((@id:[567 567])|(@id:[890 890])))))", "SORTBY", "id", "LIMIT", "0", "2"},
},
{
input: "../../tests/state/query/q7.json",
query: []interface{}{"((@id:[-inf (123])|((@org:[2 +inf]) (((@id:[567 567])|(@id:[890 890])))))", "SORTBY", "id", "LIMIT", "0", "2"},
},
}
for _, test := range tests {
data, err := os.ReadFile(test.input)

8
tests/config/state/redis/v6/statestore.yaml
View File

@ -18,6 +18,14 @@ spec:
{
"key": "message",
"type": "TEXT"
},
{
"key": "product.value",
"type": "NUMERIC"
},
{
"key": "status",
"type": "TEXT"
}
]
}

89
tests/conformance/state/state.go
View File

@ -40,6 +40,13 @@ type ValueType struct {
Message string `json:"message"`
}
type StructType struct {
Product struct {
Value int `json:"value"`
} `json:"product"`
Status string `json:"status"`
}
type intValueType struct {
Message int32 `json:"message"`
}
@ -119,6 +126,20 @@ func ConformanceTests(t *testing.T, props map[string]string, statestore state.St
value: ValueType{Message: fmt.Sprintf("test%s", key)},
contentType: contenttype.JSONContentType,
},
{
key: fmt.Sprintf("%s-struct-operations", key),
value: StructType{Product: struct {
Value int `json:"value"`
}{Value: 15}, Status: "ACTIVE"},
contentType: contenttype.JSONContentType,
},
{
key: fmt.Sprintf("%s-struct-operations-inactive", key),
value: StructType{Product: struct {
Value int `json:"value"`
}{Value: 12}, Status: "INACTIVE"},
contentType: contenttype.JSONContentType,
},
{
key: fmt.Sprintf("%s-struct-with-int", key),
value: intValueType{Message: 42},
@ -235,6 +256,67 @@ func ConformanceTests(t *testing.T, props map[string]string, statestore state.St
},
},
},
{
query: `
{
"filter": {
"AND": [
{
"GTE": {"product.value": 10}
},
{
"LT": {"product.value": 20}
},
{
"NEQ": {"status": "INACTIVE"}
}
]
}
}
`,
results: []state.QueryItem{
{
Key: fmt.Sprintf("%s-struct-operations", key),
Data: []byte(fmt.Sprintf(`{"product":{"value":15}, "status":"ACTIVE"}`)),
},
},
},
{
query: `
{
"filter": {
"OR": [
{
"AND": [
{
"GT": {"product.value": 11.1}
},
{
"EQ": {"status": "INACTIVE"}
}
]
},
{
"AND": [
{
"LTE": {"product.value": 0.5}
},
{
"EQ": {"status": "ACTIVE"}
}
]
}
]
}
}
`,
results: []state.QueryItem{
{
Key: fmt.Sprintf("%s-struct-operations-inactive", key),
Data: []byte(fmt.Sprintf(`{"product":{"value":12}, "status":"INACTIVE"}`)),
},
},
},
}
t.Run("init", func(t *testing.T) {
@ -312,6 +394,7 @@ func ConformanceTests(t *testing.T, props map[string]string, statestore state.St
metadata.ContentType: contenttype.JSONContentType,
metadata.QueryIndexName: "qIndx",
}
resp, err := querier.Query(context.Background(), &req)
require.NoError(t, err)
assert.Equal(t, len(scenario.results), len(resp.Results))
@ -1241,6 +1324,12 @@ func assertDataEquals(t *testing.T, expect any, actual []byte) {
assert.Failf(t, "unmarshal error", "error: %v, json: %s", err, string(actual))
}
assert.Equal(t, expect, v)
case StructType:
// Custom type requires case mapping
if err := json.Unmarshal(actual, &v); err != nil {
assert.Failf(t, "unmarshal error", "error: %v, json: %s", err, string(actual))
}
assert.Equal(t, expect, v)
case int:
// json.Unmarshal to float64 by default, case mapping to int coerces to int type
if err := json.Unmarshal(actual, &v); err != nil {

37
tests/state/query/q4-notequal.json Normal file
View File

@ -0,0 +1,37 @@
{
"filter": {
"OR": [
{
"EQ": {
"person.org": "A"
}
},
{
"AND": [
{
"NEQ": {
"person.org": "B"
}
},
{
"IN": {
"state": ["CA", "WA"]
}
}
]
}
]
},
"sort": [
{
"key": "state",
"order": "DESC"
},
{
"key": "person.name"
}
],
"page": {
"limit": 2
}
}

33
tests/state/query/q6-notequal.json Normal file
View File

@ -0,0 +1,33 @@
{
"filter": {
"OR": [
{
"EQ": {
"person.id": 123
}
},
{
"AND": [
{
"NEQ": {
"person.org": "B"
}
},
{
"IN": {
"person.id": [567, 890]
}
}
]
}
]
},
"sort": [
{
"key": "person.id"
}
],
"page": {
"limit": 2
}
}

36
tests/state/query/q7.json Normal file
View File

@ -0,0 +1,36 @@
{
"filter": {
"OR": [
{
"LT": {
"person.id": 123
}
},
{
"AND": [
{
"GTE": {
"person.org": 2
}
},
{
"IN": {
"person.id": [
567,
890
]
}
}
]
}
]
},
"sort": [
{
"key": "person.id"
}
],
"page": {
"limit": 2
}
}

40
tests/state/query/q8.json Normal file
View File

@ -0,0 +1,40 @@
{
"filter": {
"OR": [
{
"GTE": {
"person.org": 123
}
},
{
"AND": [
{
"LT": {
"person.org": 10
}
},
{
"IN": {
"state": [
"CA",
"WA"
]
}
}
]
}
]
},
"sort": [
{
"key": "state",
"order": "DESC"
},
{
"key": "person.name"
}
],
"page": {
"limit": 2
}
}