feat(state) - Query filter fields in the response item

.github/infrastructure/docker-compose-mongodb.yml

@@ -0,0 +1,34 @@
+version: '2'
+
+services:
+  mongodb:
+    image : mongo:4.2
+    container_name: mongodb
+    hostname: mongodb
+    restart: always
+    environment:
+      - PUID=1000
+      - PGID=1000
+      - MONGO_INITDB_DATABASE=daprStore
+      - MONGO_REPLICA_SET_NAME=test-rs
+    ports:
+      - 27017:27017
+    volumes:
+      - mongo:/data/db
+    entrypoint: [ "/usr/bin/mongod", "--bind_ip_all", "--replSet", "test-rs" ]
+
+  mongosetup:
+    image: mongo:4.2
+    links:
+      - mongodb
+    depends_on:
+      - mongodb
+    volumes:
+      - ./../scripts/components-scripts/conformance-state.mongodb-setup.sh:/scripts/mongo_setup.sh
+    restart: "no"
+    entrypoint: [ "bash", "/scripts/mongo_setup.sh" ]
+
+volumes:
+  mongo:
+
+

.github/infrastructure/docker-compose-redis7stack.yml

@@ -0,0 +1,9 @@
+version: '2'
+services:
+  redis:
+    image: redis/redis-stack:7.2.0-v11
+    ports:
+      - "6381:6379"
+      - "8001:8001"
+    environment:
+      - REDIS_REPLICATION_MODE=master

.github/scripts/components-scripts/conformance-state.mongodb-setup.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+sleep 5
+echo SETUP.sh time now: `date +"%T" `
+mongo --host mongodb:27017 <<EOF
+  var cfg = {
+    "_id": "test-rs",
+    "version": 1,
+    "members": [
+      {
+        "_id": 0,
+        "host": "localhost:27017"
+      }
+    ]
+  };
+  rs.initiate(cfg, { force: true });
+  while (! db.isMaster().ismaster ) { sleep(1000) };
+  db.getMongo().setReadPref('primary');
+  rs.status();
+EOF
+

.github/scripts/test-info.mjs

@@ -778,6 +778,11 @@ const components = {
         conformanceSetup: 'docker-compose.sh redis7 redis',
         sourcePkg: ['state/redis', 'common/component/redis'],
     },
+    'state.redis.v7stack': {
+        conformance: true,
+        conformanceSetup: 'docker-compose.sh redis7stack redis',
+        sourcePkg: ['state/redis', 'common/component/redis'],
+    },
     'state.rethinkdb': {
         conformance: true,
         conformanceSetup: 'docker-compose.sh rethinkdb',

common/component/postgresql/v1/postgresql_query.go

@@ -16,10 +16,14 @@ package postgresql
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strconv"
 	"strings"
 
+	daprmetadata "github.com/dapr/components-contrib/metadata"
+	"github.com/dapr/components-contrib/state/utils"
+
 	pginterfaces "github.com/dapr/components-contrib/common/component/postgresql/interfaces"
 	"github.com/dapr/components-contrib/state"
 	"github.com/dapr/components-contrib/state/query"
@@ -66,6 +70,13 @@ func (p *PostgreSQLQuery) Query(parentCtx context.Context, req *state.QueryReque
 		etagColumn: p.etagColumn,
 	}
 	qbuilder := query.NewQueryBuilder(q)
+	selectedAttributes, ok := daprmetadata.TryGetQuerySelectedAttributes(req.Metadata)
+	if ok {
+		var err error
+		if q.querySelectedAttributes, err = utils.ParseQuerySelectedAttributes(selectedAttributes); err != nil {
+			return nil, fmt.Errorf("postgresql store: error parsing selected attributes: %w", err)
+		}
+	}
 	if err := qbuilder.BuildQuery(&req.Query); err != nil {
 		return &state.QueryResponse{}, err
 	}
@@ -81,12 +92,13 @@ func (p *PostgreSQLQuery) Query(parentCtx context.Context, req *state.QueryReque
 }
 
 type Query struct {
-	query      string
-	params     []interface{}
-	limit      int
-	skip       *int64
-	tableName  string
-	etagColumn string
+	query                   string
+	params                  []interface{}
+	limit                   int
+	skip                    *int64
+	tableName               string
+	etagColumn              string
+	querySelectedAttributes []utils.Attribute
 }
 
 func (q *Query) VisitEQ(f *query.EQ) (string, error) {
@@ -222,8 +234,19 @@ func (q *Query) VisitOR(f *query.OR) (string, error) {
 }
 
 func (q *Query) Finalize(filters string, qq *query.Query) error {
-	q.query = fmt.Sprintf("SELECT key, value, %s as etag FROM "+q.tableName, q.etagColumn)
+	if q.querySelectedAttributes != nil {
+		var columns string
+		for idx, item := range q.querySelectedAttributes {
+			if idx != 0 {
+				columns += ", "
+			}
+			columns += translateFieldToFilter(item.Path) + " as " + item.Name
+		}
 
+		q.query = fmt.Sprintf("SELECT key, %s, %s as etag FROM "+q.tableName, columns, q.etagColumn)
+	} else {
+		q.query = fmt.Sprintf("SELECT key, value, %s as etag FROM "+q.tableName, q.etagColumn)
+	}
 	if filters != "" {
 		q.query += " WHERE " + filters
 	}
@@ -267,14 +290,59 @@ func (q *Query) execute(ctx context.Context, db pginterfaces.DBQuerier) ([]state
 	defer rows.Close()
 
 	ret := []state.QueryItem{}
+
 	for rows.Next() {
 		var (
 			key  string
 			data []byte
 			etag uint32
 		)
-		if err = rows.Scan(&key, &data, &etag); err != nil {
-			return nil, "", err
+		if q.querySelectedAttributes != nil {
+			var values []interface{}
+			if values, err = rows.Values(); err != nil {
+				return nil, "", err
+			}
+			var ok bool
+			key = values[0].(string)
+			etag, ok = values[len(values)-1].(uint32)
+			if !ok {
+				// Cockroachdb the type etag is int64 instead of uint32, necessary the casting
+				etag = uint32(values[len(values)-1].(int64))
+			}
+			result := make(map[string]interface{})
+			for idx, item := range q.querySelectedAttributes {
+				value := values[idx+1]
+				switch item.Type {
+				case utils.Bool:
+					result[item.Name], err = strconv.ParseBool(value.(string))
+				case utils.Numeric:
+					var jsonObject interface{}
+					err = json.Unmarshal([]byte(value.(string)), &jsonObject)
+					result[item.Name] = jsonObject
+				case utils.Object:
+					var jsonObject interface{}
+					err = json.Unmarshal([]byte(value.(string)), &jsonObject)
+					result[item.Name] = jsonObject
+				case utils.Array:
+					var jsonArray []interface{}
+					err = json.Unmarshal([]byte(value.(string)), &jsonArray)
+					result[item.Name] = jsonArray
+				default:
+					result[item.Name] = value.(string)
+				}
+
+				if err != nil {
+					return nil, "", err
+				}
+				data, err = json.Marshal(result)
+				if err != nil {
+					return nil, "", err
+				}
+			}
+		} else {
+			if err = rows.Scan(&key, &data, &etag); err != nil {
+				return nil, "", err
+			}
 		}
 		result := state.QueryItem{
 			Key:  key,

common/component/postgresql/v1/postgresql_query_test.go

@@ -18,6 +18,8 @@ import (
 	"os"
 	"testing"
 
+	"github.com/dapr/components-contrib/state/utils"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -26,8 +28,9 @@ import (
 
 func TestPostgresqlQueryBuildQuery(t *testing.T) {
 	tests := []struct {
-		input string
-		query string
+		input              string
+		query              string
+		selectedAttributes string
 	}{
 		{
 			input: "../../../../tests/state/query/q1.json",
@@ -61,18 +64,30 @@ func TestPostgresqlQueryBuildQuery(t *testing.T) {
 			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",
 		},
+		{
+			input:              "../../../../tests/state/query/q8.json",
+			query:              "SELECT key, value->>'person' as Person, value->'person'->>'org' as Organization, 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",
+			selectedAttributes: `[{"name":"Person", "path":"person", "type":"Object"},{"name":"Organization", "path":"person.org", "type":"Text"}]`,
+		},
 	}
 	for _, test := range tests {
 		data, err := os.ReadFile(test.input)
 		require.NoError(t, err)
 		var qq query.Query
+
 		err = json.Unmarshal(data, &qq)
 		require.NoError(t, err)
 
 		q := &Query{
 			tableName:  defaultTableName,
 			etagColumn: "xmin",
 		}
+
+		if len(test.selectedAttributes) > 0 {
+			q.querySelectedAttributes, err = utils.ParseQuerySelectedAttributes(test.selectedAttributes)
+			require.NoError(t, err)
+		}
+
 		qbuilder := query.NewQueryBuilder(q)
 		err = qbuilder.BuildQuery(&qq)
 		require.NoError(t, err)

metadata/utils.go

@@ -42,6 +42,9 @@ const (
 	// QueryIndexName defines the metadata key for the name of query indexing schema (for redis).
 	QueryIndexName = "queryIndexName"
 
+	// QuerySelectedAttributes defines the metadata key for getting only the selected attributes in the result item (for redis)
+	QuerySelectedAttributes = "querySelectedAttributes"
+
 	// MaxBulkPubBytesKey defines the maximum bytes to publish in a bulk publish request metadata.
 	MaxBulkPubBytesKey string = "maxBulkPubBytes"
 )
@@ -130,6 +133,14 @@ func TryGetQueryIndexName(props map[string]string) (string, bool) {
 	return "", false
 }
 
+func TryGetQuerySelectedAttributes(props map[string]string) (string, bool) {
+	if val, ok := props[QuerySelectedAttributes]; ok && val != "" {
+		return val, true
+	}
+
+	return "", false
+}
+
 // GetMetadataProperty returns a property from the metadata map, with support for aliases
 func GetMetadataProperty(props map[string]string, keys ...string) (val string, ok bool) {
 	lcProps := make(map[string]string, len(props))

state/azure/cosmosdb/cosmosdb.go

@@ -592,6 +592,15 @@ func (c *StateStore) Query(ctx context.Context, req *state.QueryRequest) (*state
 	q := &Query{}
 
 	qbuilder := query.NewQueryBuilder(q)
+
+	selectedAttributes, ok := contribmeta.TryGetQuerySelectedAttributes(req.Metadata)
+	if ok {
+		var err error
+		if q.querySelectedAttributes, err = stateutils.ParseQuerySelectedAttributes(selectedAttributes); err != nil {
+			return nil, fmt.Errorf("postgresql store: error parsing selected attributes: %w", err)
+		}
+	}
+
 	if err := qbuilder.BuildQuery(&req.Query); err != nil {
 		return &state.QueryResponse{}, err
 	}

state/azure/cosmosdb/cosmosdb_query.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/dapr/components-contrib/state/utils"
+
 	"github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos"
 	jsoniter "github.com/json-iterator/go"
 
@@ -34,10 +36,11 @@ type InternalQuery struct {
 }
 
 type Query struct {
-	query        InternalQuery
-	limit        int
-	token        string
-	partitionKey string
+	query                   InternalQuery
+	limit                   int
+	token                   string
+	partitionKey            string
+	querySelectedAttributes []utils.Attribute
 }
 
 func (q *Query) VisitEQ(f *query.EQ) (string, error) {
@@ -226,7 +229,16 @@ func (q *Query) Finalize(filters string, qq *query.Query) error {
 		orderBy = " ORDER BY " + strings.Join(order, ", ")
 	}
 
-	q.query.query = "SELECT * FROM c" + filter + orderBy
+	if q.querySelectedAttributes != nil {
+		var columns string
+		columns = "c['id'], c['_etag']"
+		for _, item := range q.querySelectedAttributes {
+			columns += ", " + replaceKeywords("c.value."+item.Path) + " as '" + item.Name + "'"
+		}
+		q.query.query = "SELECT " + columns + " FROM c" + filter + orderBy
+	} else {
+		q.query.query = "SELECT * FROM c" + filter + orderBy
+	}
 	q.limit = qq.Page.Limit
 	q.token = qq.Page.Token
 
@@ -272,7 +284,6 @@ func (q *Query) execute(ctx context.Context, client *azcosmos.ContainerClient) (
 	} else {
 		pk = azcosmos.NewPartitionKeyBool(true)
 	}
-
 	queryPager := client.NewQueryItemsPager(q.query.query, pk, opts)
 
 	token := ""
@@ -291,7 +302,23 @@ func (q *Query) execute(ctx context.Context, client *azcosmos.ContainerClient) (
 		}
 		for _, item := range queryResponse.Items {
 			tempItem := CosmosItem{}
-			err := json.Unmarshal(item, &tempItem)
+			var err error
+			if q.querySelectedAttributes != nil {
+				properties := make(map[string]interface{})
+				err = json.Unmarshal(item, &properties)
+				tempItem.ID = properties["id"].(string)
+				tempItem.Etag = properties["_etag"].(string)
+				if err != nil {
+					return nil, "", err
+				}
+				value := make(map[string]interface{})
+				for _, item := range q.querySelectedAttributes {
+					value[item.Name] = properties[item.Name]
+				}
+				tempItem.Value = value
+			} else {
+				err = json.Unmarshal(item, &tempItem)
+			}
 			if err != nil {
 				return nil, "", err
 			}