1// Copyright 2013-2020 Aerospike, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15package aerospike_test
16
17import (
18	"fmt"
19
20	as "github.com/aerospike/aerospike-client-go"
21
22	. "github.com/onsi/ginkgo"
23	. "github.com/onsi/gomega"
24)
25
26// ALL tests are isolated by SetName and Key, which are 50 random characters
27var _ = Describe("Query operations on complex types", func() {
28
29	// connection data
30	var ns = *namespace
31	var set = randString(50)
32	var wpolicy = as.NewWritePolicy(0, 0)
33	wpolicy.SendKey = true
34
35	const keyCount = 1000
36
37	valueList := []interface{}{1, 2, 3, "a", "ab", "abc"}
38	valueMap := map[interface{}]interface{}{"a": "b", 0: 1, 1: "a", "b": 2}
39
40	bin1 := as.NewBin("List", valueList)
41	bin2 := as.NewBin("Map", valueMap)
42	var keys map[string]*as.Key
43
44	BeforeEach(func() {
45		keys = make(map[string]*as.Key, keyCount)
46		set = randString(50)
47		for i := 0; i < keyCount; i++ {
48			key, err := as.NewKey(ns, set, randString(50))
49			Expect(err).ToNot(HaveOccurred())
50
51			keys[string(key.Digest())] = key
52			err = client.PutBins(wpolicy, key, bin1, bin2)
53			Expect(err).ToNot(HaveOccurred())
54		}
55
56		// queries only work on indices
57		idxTask1, err := client.CreateComplexIndex(wpolicy, ns, set, set+bin1.Name, bin1.Name, as.NUMERIC, as.ICT_LIST)
58		Expect(err).ToNot(HaveOccurred())
59
60		// wait until index is created
61		Expect(<-idxTask1.OnComplete()).ToNot(HaveOccurred())
62
63		// queries only work on indices
64		idxTask2, err := client.CreateComplexIndex(wpolicy, ns, set, set+bin2.Name+"keys", bin2.Name, as.NUMERIC, as.ICT_MAPKEYS)
65		Expect(err).ToNot(HaveOccurred())
66
67		// wait until index is created
68		Expect(<-idxTask2.OnComplete()).ToNot(HaveOccurred())
69
70		// queries only work on indices
71		idxTask3, err := client.CreateComplexIndex(wpolicy, ns, set, set+bin2.Name+"values", bin2.Name, as.NUMERIC, as.ICT_MAPVALUES)
72		Expect(err).ToNot(HaveOccurred())
73
74		// wait until index is created
75		Expect(<-idxTask3.OnComplete()).ToNot(HaveOccurred())
76	})
77
78	AfterEach(func() {
79		Expect(client.DropIndex(nil, ns, set, set+bin1.Name)).ToNot(HaveOccurred())
80		Expect(client.DropIndex(nil, ns, set, set+bin2.Name+"keys")).ToNot(HaveOccurred())
81		Expect(client.DropIndex(nil, ns, set, set+bin2.Name+"values")).ToNot(HaveOccurred())
82	})
83
84	for _, failOnClusterChange := range []bool{false, true} {
85		var queryPolicy = as.NewQueryPolicy()
86		queryPolicy.FailOnClusterChange = failOnClusterChange
87
88		It(fmt.Sprintf("must Query a specific element in list and get only relevant records back. FailOnClusterChange: %v", failOnClusterChange), func() {
89			stm := as.NewStatement(ns, set)
90			stm.SetFilter(as.NewContainsFilter(bin1.Name, as.ICT_LIST, 1))
91			recordset, err := client.Query(queryPolicy, stm)
92			Expect(err).ToNot(HaveOccurred())
93
94			cnt := 0
95			for res := range recordset.Results() {
96				Expect(res.Err).ToNot(HaveOccurred())
97				rec := res.Record
98				cnt++
99				_, exists := keys[string(rec.Key.Digest())]
100				Expect(exists).To(Equal(true))
101			}
102
103			Expect(cnt).To(BeNumerically("==", keyCount))
104		})
105
106		It(fmt.Sprintf("must Query a specific non-existig element in list and get no records back. FailOnClusterChange: %v", failOnClusterChange), func() {
107			stm := as.NewStatement(ns, set)
108			stm.SetFilter(as.NewContainsFilter(bin1.Name, as.ICT_LIST, 10))
109			recordset, err := client.Query(queryPolicy, stm)
110			Expect(err).ToNot(HaveOccurred())
111
112			cnt := 0
113			for res := range recordset.Results() {
114				Expect(res.Err).ToNot(HaveOccurred())
115				cnt++
116			}
117
118			Expect(cnt).To(BeNumerically("==", 0))
119		})
120
121		It(fmt.Sprintf("must Query a key in map and get only relevant records back. FailOnClusterChange: %v", failOnClusterChange), func() {
122			stm := as.NewStatement(ns, set)
123			stm.SetFilter(as.NewContainsFilter(bin2.Name, as.ICT_MAPKEYS, 0))
124			recordset, err := client.Query(queryPolicy, stm)
125			Expect(err).ToNot(HaveOccurred())
126
127			cnt := 0
128			for res := range recordset.Results() {
129				Expect(res.Err).ToNot(HaveOccurred())
130				rec := res.Record
131				cnt++
132				_, exists := keys[string(rec.Key.Digest())]
133				Expect(exists).To(Equal(true))
134			}
135
136			Expect(cnt).To(BeNumerically("==", keyCount))
137		})
138
139		It(fmt.Sprintf("must Query a specific non-existig key in map and get no records back. FailOnClusterChange: %v", failOnClusterChange), func() {
140			stm := as.NewStatement(ns, set)
141			stm.SetFilter(as.NewContainsFilter(bin2.Name, as.ICT_MAPKEYS, 10))
142			recordset, err := client.Query(queryPolicy, stm)
143			Expect(err).ToNot(HaveOccurred())
144
145			cnt := 0
146			for res := range recordset.Results() {
147				Expect(res.Err).ToNot(HaveOccurred())
148				cnt++
149			}
150
151			Expect(cnt).To(BeNumerically("==", 0))
152		})
153
154		It(fmt.Sprintf("must Query a value in map and get only relevant records back. FailOnClusterChange: %v", failOnClusterChange), func() {
155			stm := as.NewStatement(ns, set)
156			stm.SetFilter(as.NewContainsFilter(bin2.Name, as.ICT_MAPVALUES, 1))
157			recordset, err := client.Query(queryPolicy, stm)
158			Expect(err).ToNot(HaveOccurred())
159
160			cnt := 0
161			for res := range recordset.Results() {
162				Expect(res.Err).ToNot(HaveOccurred())
163				rec := res.Record
164				cnt++
165				_, exists := keys[string(rec.Key.Digest())]
166				Expect(exists).To(Equal(true))
167			}
168
169			Expect(cnt).To(BeNumerically("==", keyCount))
170		})
171
172		It(fmt.Sprintf("must Query a specific non-existig value in map and get no records back. FailOnClusterChange: %v", failOnClusterChange), func() {
173			stm := as.NewStatement(ns, set)
174			stm.SetFilter(as.NewContainsFilter(bin2.Name, as.ICT_MAPVALUES, 10))
175			recordset, err := client.Query(queryPolicy, stm)
176			Expect(err).ToNot(HaveOccurred())
177
178			cnt := 0
179			for res := range recordset.Results() {
180				Expect(res.Err).ToNot(HaveOccurred())
181				cnt++
182			}
183
184			Expect(cnt).To(BeNumerically("==", 0))
185		})
186	}
187})
188