Fix Redis Query for RediSearch 2.8+ (dapr#3303)

Signed-off-by: joshvanl <[email protected]>
Co-authored-by: Alessandro (Ale) Segala <[email protected]>
Signed-off-by: Alessandro (Ale) Segala <[email protected]>

Author: JoshVanL

state/redis/redis_query.go

@@ -295,20 +295,83 @@ func (q *Query) execute(ctx context.Context, client rediscomponent.RedisClient)
 	if err != nil {
 		return nil, "", err
 	}
-	arr, ok := ret.([]interface{})
+
+	res, ok, err := parseQueryResponsePost28(ret)
+	if err != nil {
+		return nil, "", err
+	}
+	if !ok {
+		res, err = parseQueryResponsePre28(ret)
+		if err != nil {
+			return nil, "", err
+		}
+	}
+
+	// set next query token only if limit is specified
+	var token string
+	if q.limit > 0 && len(res) > 0 {
+		token = strconv.FormatInt(q.offset+int64(len(res)), 10)
+	}
+
+	return res, token, err
+}
+
+// parseQueryResponsePost28 parses the query Do response from redisearch 2.8+.
+func parseQueryResponsePost28(ret any) ([]state.QueryItem, bool, error) {
+	aarr, ok := ret.(map[any]any)
 	if !ok {
-		return nil, "", fmt.Errorf("invalid output")
+		return nil, false, nil
+	}
+
+	var res []state.QueryItem
+	arr := aarr["results"].([]any)
+	if len(arr) == 0 {
+		return nil, false, errors.New("invalid output")
 	}
-	// arr[0] = number of matching elements in DB (ignoring pagination)
+	for i := 0; i < len(arr); i++ {
+		inner, ok := arr[i].(map[any]any)
+		if !ok {
+			return nil, false, fmt.Errorf("invalid output")
+		}
+		exattr, ok := inner["extra_attributes"].(map[any]any)
+		if !ok {
+			return nil, false, fmt.Errorf("invalid output")
+		}
+		item := state.QueryItem{
+			Key: inner["id"].(string),
+		}
+		if data, ok := exattr["$.data"].(string); ok {
+			item.Data = []byte(data)
+		} else {
+			item.Error = fmt.Sprintf("%#v is not string", exattr["$.data"])
+		}
+		if etag, ok := exattr["$.version"].(string); ok {
+			item.ETag = &etag
+		}
+		res = append(res, item)
+	}
+
+	return res, true, nil
+}
+
+// parseQueryResponsePre28 parses the query Do response from redisearch 2.8-.
+func parseQueryResponsePre28(ret any) ([]state.QueryItem, error) {
+	arr, ok := ret.([]any)
+	if !ok {
+		return nil, errors.New("invalid output")
+	}
+
+	// arr[0] = number of matching elements in DB (ignoring pagination
 	// arr[2n] = key
 	// arr[2n+1][0] = "$.data"
 	// arr[2n+1][1] = value
 	// arr[2n+1][2] = "$.version"
 	// arr[2n+1][3] = etag
 	if len(arr)%2 != 1 {
-		return nil, "", fmt.Errorf("invalid output")
+		return nil, fmt.Errorf("invalid output")
 	}
-	res := []state.QueryItem{}
+
+	var res []state.QueryItem
 	for i := 1; i < len(arr); i += 2 {
 		item := state.QueryItem{
 			Key: arr[i].(string),
@@ -322,11 +385,6 @@ func (q *Query) execute(ctx context.Context, client rediscomponent.RedisClient)
 		}
 		res = append(res, item)
 	}
-	// set next query token only if limit is specified
-	var token string
-	if q.limit > 0 && len(res) > 0 {
-		token = strconv.FormatInt(q.offset+int64(len(res)), 10)
-	}
 
-	return res, token, err
+	return res, nil
 }

tests/certification/state/redis/components/docker/default/redisstatestore.yaml

@@ -13,3 +13,16 @@ spec:
     value: 5m
   - name: timeout
     value: 100s
+  - name: queryIndexes
+    value: |
+      [
+        {
+          "name": "tquery",
+          "indexes": [
+            {
+              "key": "qmsg",
+              "type": "TEXT"
+            }
+          ]
+        }
+      ]

tests/certification/state/redis/docker-compose.yml

@@ -1,7 +1,8 @@
 version: "3.3"
 services:
   redis:
+    healthcheck:
+      test: ["CMD", "redis-cli","ping"]
     image: 'redislabs/redisearch:latest'
     ports:
       - '6379:6379'
-    command: redis-server

tests/certification/state/redis/redis_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/dapr/components-contrib/state"
+	"github.com/dapr/components-contrib/state/query"
 	state_redis "github.com/dapr/components-contrib/state/redis"
 	"github.com/dapr/components-contrib/tests/certification/embedded"
 	"github.com/dapr/components-contrib/tests/certification/flow"
@@ -248,12 +249,74 @@ func TestRedis(t *testing.T) {
 		return nil
 	}
 
+	type qValue struct {
+		Qmsg string `json:"qmsg"`
+	}
+
+	// Query test
+	queryTest := func(ctx flow.Context) error {
+		err := stateStore.Multi(context.Background(), &state.TransactionalStateRequest{
+			Operations: []state.TransactionalStateOperation{
+				state.SetRequest{
+					Key:   "qKey1",
+					Value: qValue{Qmsg: "test1"},
+				},
+				state.SetRequest{
+					Key:   "qKey2",
+					Value: qValue{Qmsg: "test2"},
+				},
+				state.SetRequest{
+					Key:   "qKey3",
+					Value: qValue{Qmsg: "test3"},
+				},
+			},
+			Metadata: map[string]string{
+				"contentType": "application/json",
+			},
+		})
+		require.NoError(t, err)
+		resp, err := stateStore.Query(context.Background(), &state.QueryRequest{
+			Query: query.Query{
+				QueryFields: query.QueryFields{
+					Filters: map[string]any{
+						"OR": []any{
+							map[string]any{
+								"EQ": map[string]any{
+									"qmsg": "test1",
+								},
+							},
+							map[string]any{
+								"EQ": map[string]any{
+									"qmsg": "test2",
+								},
+							},
+						},
+					},
+				},
+				Filter: &query.OR{
+					Filters: []query.Filter{
+						&query.EQ{Key: "qmsg", Val: "test1"},
+						&query.EQ{Key: "qmsg", Val: "test2"},
+					},
+				},
+			},
+			Metadata: map[string]string{
+				"queryIndexName": "tquery",
+				"contentType":    "application/json",
+			},
+		})
+		require.NoError(t, err)
+		require.Len(t, resp.Results, 2)
+		assert.Equal(t, "qKey1", resp.Results[0].Key)
+		assert.Equal(t, "qKey2", resp.Results[1].Key)
+		assert.JSONEq(t, `{"qmsg":"test1"}`, string(resp.Results[0].Data))
+		assert.JSONEq(t, `{"qmsg":"test2"}`, string(resp.Results[1].Data))
+		return nil
+	}
+
 	testForStateStoreNotConfigured := func(ctx flow.Context) error {
 		client, err := client.NewClientWithPort(fmt.Sprint(currentGrpcPort))
-		// require.Error(t, err)
-		if err != nil {
-			panic(err)
-		}
+		require.NoError(t, err)
 		defer client.Close()
 
 		err = client.SaveState(ctx, stateStoreName, certificationTestPrefix+"key1", []byte("redisCert"), nil)
@@ -289,6 +352,7 @@ func TestRedis(t *testing.T) {
 			embedded.WithStates(stateRegistry),
 		)).
 		Step("Run basic test", basicTest).
+		Step("Run test for Query", queryTest).
 		Step("Run TTL related test", timeToLiveTest).
 		Step("interrupt network",
 			network.InterruptNetwork(10*time.Second, nil, nil, "6379:6379")).