go/src/infra/gae/libs/wrapper/memory/datastore_query.go - Issue 1222903002: Refactor current GAE abstraction library to be free of the SDK*

Side by Side Diff: go/src/infra/gae/libs/wrapper/memory/datastore_query.go

Issue 1222903002: Refactor current GAE abstraction library to be free of the SDK* (Closed) Base URL: https://chromium.googlesource.com/infra/infra.git@master

Patch Set: more fixes Created 5 years, 5 months ago

Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.

Jump to:

View unified diff | Download patch

« go/src/infra/gae/libs/gae/prod/datastore_key.go ('K') | « go/src/infra/gae/libs/wrapper/memory/datastore_data.go ('k') | go/src/infra/gae/libs/wrapper/memory/datastore_test.go » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Hide Comments ('s')

OLD	NEW
	(Empty)
1 // Copyright 2015 The Chromium Authors. All rights reserved.

2 // Use of this source code is governed by a BSD-style license that can be

3 // found in the LICENSE file.

4

5 package memory

6

7 import (

8 "bytes"

9 "errors"

10 "fmt"

11 "math"

12 "strings"

13

14 "appengine/datastore"

15 pb "appengine_internal/datastore"

16

17 "github.com/luci/gkvlite"

18 "github.com/luci/luci-go/common/cmpbin"

19

20 "infra/gae/libs/wrapper"

21 )

22

23 type qDirection bool

24

25 const (

26 qASC qDirection = true

27 qDEC = false

28 )

29

30 var builtinQueryPrefix = []byte{0}

31 var complexQueryPrefix = []byte{1}

32

33 type qSortBy struct {

34 prop string

35 dir qDirection

36 }

37

38 func (q qSortBy) WriteBinary(buf *bytes.Buffer) {

39 if q.dir == qASC {

40 buf.WriteByte(0)

41 } else {

42 buf.WriteByte(1)

43 }

44 writeString(buf, q.prop)

45 }

46

47 func (q qSortBy) ReadBinary(buf bytes.Buffer) error {

48 dir, err := buf.ReadByte()

49 if err != nil {

50 return err

51 }

52 q.dir = dir == 0

53 q.prop, err = readString(buf)

54 return err

55 }

56

57 type qIndex struct {

58 kind string

59 ancestor bool

60 sortby []qSortBy

61 }

62

63 func (i *qIndex) Builtin() bool {

64 return !i.ancestor && len(i.sortby) <= 1

65 }

66

67 // Valid verifies that this qIndex doesn't have duplicate sortBy fields.

68 func (i *qIndex) Valid() bool {

69 names := map[string]bool{}

70 for _, sb := range i.sortby {

71 if names[sb.prop] {

72 return false

73 }

74 names[sb.prop] = true

75 }

76 return true

77 }

78

79 func (i qIndex) WriteBinary(buf bytes.Buffer) {

80 // TODO(riannucci): do a Grow call here?

81 if i.Builtin() {

82 buf.Write(builtinQueryPrefix)

83 } else {

84 buf.Write(complexQueryPrefix)

85 }

86 writeString(buf, i.kind)

87 if i.ancestor {

88 buf.WriteByte(0)

89 } else {

90 buf.WriteByte(1)

91 }

92 cmpbin.WriteUint(buf, uint64(len(i.sortby)))

93 for _, sb := range i.sortby {

94 sb.WriteBinary(buf)

95 }

96 }

97

98 func (i *qIndex) String() string {

99 ret := &bytes.Buffer{}

100 if i.Builtin() {

101 ret.WriteRune('B')

102 } else {

103 ret.WriteRune('C')

104 }

105 ret.WriteRune(':')

106 ret.WriteString(i.kind)

107 if i.ancestor {

108 ret.WriteString("\|A")

109 }

110 for _, sb := range i.sortby {

111 ret.WriteRune('/')

112 if sb.dir == qDEC {

113 ret.WriteRune('-')

114 }

115 ret.WriteString(sb.prop)

116 }

117 return ret.String()

118 }

119

120 func (i qIndex) ReadBinary(buf bytes.Buffer) error {

121 // discard builtin/complex byte

122 _, err := buf.ReadByte()

123 if err != nil {

124 return err

125 }

126

127 i.kind, err = readString(buf)

128 if err != nil {

129 return err

130 }

131 anc, err := buf.ReadByte()

132 if err != nil {

133 return err

134 }

135 i.ancestor = anc == 1

136

137 numSorts, _, err := cmpbin.ReadUint(buf)

138 if err != nil {

139 return err

140 }

141 if numSorts > 64 {

142 return fmt.Errorf("qIndex.ReadBinary: Got over 64 sort orders: % d", numSorts)

143 }

144 i.sortby = make([]qSortBy, numSorts)

145 for idx := range i.sortby {

146 err = (&i.sortby[idx]).ReadBinary(buf)

147 if err != nil {

148 return err

149 }

150 }

151

152 return nil

153 }

154

155 type queryOp int

156

157 const (

158 qInvalid queryOp = iota

159 qEqual

160 qLessThan

161 qLessEq

162 qGreaterEq

163 qGreaterThan

164 )

165

166 func (o queryOp) isEQOp() bool {

167 return o == qEqual

168 }

169

170 func (o queryOp) isINEQOp() bool {

171 return o >= qLessThan && o <= qGreaterThan

172 }

173

174 var queryOpMap = map[string]queryOp{

175 "=": qEqual,

176 "<": qLessThan,

177 "<=": qLessEq,

178 ">=": qGreaterEq,

179 ">": qGreaterThan,

180 }

181

182 type queryFilter struct {

183 field string

184 op queryOp

185 value interface{}

186 }

187

188 func parseFilter(f string, v interface{}) (ret queryFilter, err error) {

189 toks := strings.SplitN(strings.TrimSpace(f), " ", 2)

190 if len(toks) != 2 {

191 err = errors.New("datastore: invalid filter: " + f)

192 } else {

193 op := queryOpMap[toks[1]]

194 if op == qInvalid {

195 err = fmt.Errorf("datastore: invalid operator %q in filt er %q", toks[1], f)

196 } else {

197 ret.field = toks[0]

198 ret.op = op

199 ret.value = v

200 }

201 }

202 return

203 }

204

205 type queryOrder struct {

206 field string

207 direction qDirection

208 }

209

210 type queryCursor string

211

212 func (q queryCursor) String() string { return string(q) }

213 func (q queryCursor) Valid() bool { return q != "" }

214

215 type queryImpl struct {

216 wrapper.DSQuery

217

218 ns string

219

220 kind string

221 ancestor *datastore.Key

222 filter []queryFilter

223 order []queryOrder

224

225 keysOnly bool

226 limit int32

227 offset int32

228

229 start queryCursor

230 end queryCursor

231

232 err error

233 }

234

235 type queryIterImpl struct {

236 idx *queryImpl

237 }

238

239 func (q *queryIterImpl) Cursor() (wrapper.DSCursor, error) {

240 if q.idx.err != nil {

241 return nil, q.idx.err

242 }

243 return nil, nil

244 }

245

246 func (q queryIterImpl) Next(dst interface{}) (datastore.Key, error) {

247 if q.idx.err != nil {

248 return nil, q.idx.err

249 }

250 return nil, nil

251 }

252

253 func (q queryImpl) normalize() (ret queryImpl) {

254 // ported from GAE SDK datastore_index.py;Normalize()

255 ret = q.clone()

256

257 bs := newMemStore()

258

259 eqProperties := bs.MakePrivateCollection(nil)

260

261 ineqProperties := bs.MakePrivateCollection(nil)

262

263 for _, f := range ret.filter {

264 // if we supported the IN operator, we would check to see if the re were

265 // multiple value operands here, but the go SDK doesn't support this.

266 if f.op.isEQOp() {

267 eqProperties.Set([]byte(f.field), []byte{})

268 } else if f.op.isINEQOp() {

269 ineqProperties.Set([]byte(f.field), []byte{})

270 }

271 }

272

273 ineqProperties.VisitItemsAscend(nil, false, func(i *gkvlite.Item) bool {

274 eqProperties.Delete(i.Key)

275 return true

276 })

277

278 removeSet := bs.MakePrivateCollection(nil)

279 eqProperties.VisitItemsAscend(nil, false, func(i *gkvlite.Item) bool {

280 removeSet.Set(i.Key, []byte{})

281 return true

282 })

283

284 newOrders := []queryOrder{}

285 for _, o := range ret.order {

286 if removeSet.Get([]byte(o.field)) == nil {

287 removeSet.Set([]byte(o.field), []byte{})

288 newOrders = append(newOrders, o)

289 }

290 }

291 ret.order = newOrders

292

293 // need to fix ret.filters if we ever support the EXISTS operator and/or

294 // projections.

295 //

296 // newFilters = []

297 // for f in ret.filters:

298 // if f.op != qExists:

299 // newFilters = append(newFilters, f)

300 // if !removeSet.Has(f.field):

301 // removeSet.InsertNoReplace(f.field)

302 // newFilters = append(newFilters, f)

303 //

304 // so ret.filters == newFilters becuase none of ret.filters has op == qE xists

305 //

306 // then:

307 //

308 // for prop in ret.project:

309 // if !removeSet.Has(prop):

310 // removeSet.InsertNoReplace(prop)

311 // ... make new EXISTS filters, add them to newFilters ...

312 // ret.filters = newFilters

313 //

314 // However, since we don't support projection queries, this is moot.

315

316 if eqProperties.Get([]byte("__key__")) != nil {

317 ret.order = []queryOrder{}

318 }

319

320 newOrders = []queryOrder{}

321 for _, o := range ret.order {

322 if o.field == "__key__" {

323 newOrders = append(newOrders, o)

324 break

325 }

326 newOrders = append(newOrders, o)

327 }

328 ret.order = newOrders

329

330 return

331 }

332

333 func (q queryImpl) checkCorrectness(ns string, isTxn bool) (ret queryImpl) {

334 // ported from GAE SDK datastore_stub_util.py;CheckQuery()

335 ret = q.clone()

336

337 if ns != ret.ns {

338 ret.err = newDSError(pb.Error_BAD_REQUEST,

339 "MADE UP ERROR: Namespace mismatched. Query and Datastor e don't agree "+

340 "on the current namespace")

341 return

342 }

343

344 if ret.err != nil {

345 return

346 }

347

348 // if projection && keys_only:

349 // "projection and keys_only cannot both be set"

350

351 // if projection props match /^__.*__$/:

352 // "projections are not supported for the property: %(prop)s"

353

354 if isTxn && ret.ancestor == nil {

355 ret.err = newDSError(pb.Error_BAD_REQUEST,

356 "Only ancestor queries are allowed inside transactions")

357 return

358 }

359

360 numComponents := len(ret.filter) + len(ret.order)

361 if ret.ancestor != nil {

362 numComponents++

363 }

364 if numComponents > 100 {

365 ret.err = newDSError(pb.Error_BAD_REQUEST,

366 "query is too large. may not have more than "+

367 "100 filters + sort orders ancestor total")

368 }

369

370 // if ret.ancestor.appid() != current appid

371 // "query app is x but ancestor app is x"

372 // if ret.ancestor.namespace() != current namespace

373 // "query namespace is x but ancestor namespace is x"

374

375 // if not all(g in orders for g in group_by)

376 // "items in the group by clause must be specified first in the orderin g"

377

378 ineqPropName := ""

379 for _, f := range ret.filter {

380 if f.field == "__key__" {

381 k, ok := f.value.(*datastore.Key)

382 if !ok {

383 ret.err = newDSError(pb.Error_BAD_REQUEST,

384 "__key__ filter value must be a Key")

385 return

386 }

387 if !keyValid(ret.ns, k, userKeyOnly) {

388 // See the comment in queryImpl.Ancestor; basica lly this check

389 // never happens in the real env because the SDK silently swallows

390 // this condition :/

391 ret.err = datastore.ErrInvalidKey

392 return

393 }

394 // __key__ filter app is X but query app is X

395 // __key__ filter namespace is X but query namespace is X

396 }

397 // if f.op == qEqual and f.field in ret.project_fields

398 // "cannot use projection on a proprety with an equality filte r"

399

400 if f.op.isINEQOp() {

401 if ineqPropName == "" {

402 ineqPropName = f.field

403 } else if f.field != ineqPropName {

404 ret.err = newDSError(pb.Error_BAD_REQUEST,

405 fmt.Sprintf(

406 "Only one inequality filter per query is supported. "+

407 "Encountered both %s and %s", ineqPropName, f.field))

408 return

409 }

410 }

411 }

412

413 // if ineqPropName != "" && len(group_by) > 0 && len(orders) ==0

414 // "Inequality filter on X must also be a group by property "+

415 // "when group by properties are set."

416

417 if ineqPropName != "" && len(ret.order) != 0 {

418 if ret.order[0].field != ineqPropName {

419 ret.err = newDSError(pb.Error_BAD_REQUEST,

420 fmt.Sprintf(

421 "The first sort property must be the sam e as the property "+

422 "to which the inequality filter is applied. In your query "+

423 "the first sort property is %s b ut the inequality filter "+

424 "is on %s", ret.order[0].field, ineqPropName))

425 return

426 }

427 }

428

429 if ret.kind == "" {

430 for _, f := range ret.filter {

431 if f.field != "__key__" {

432 ret.err = newDSError(pb.Error_BAD_REQUEST,

433 "kind is required for non-__key__ filter s")

434 return

435 }

436 }

437 for _, o := range ret.order {

438 if o.field != "__key__" \|\| o.direction != qASC {

439 ret.err = newDSError(pb.Error_BAD_REQUEST,

440 "kind is required for all orders except __key__ ascending")

441 return

442 }

443 }

444 }

445 return

446 }

447

448 func (q queryImpl) calculateIndex() qIndex {

449 // as a nod to simplicity in this code, we'll require that a single inde x

450 // is able to service the entire query. E.g. no zigzag merge joins or

451 // multiqueries. This will mean that the user will need to rely on

452 // dev_appserver to tell them what indicies they need for real, and for thier

453 // tests they'll need to specify the missing composite indices manually.

454 //

455 // This COULD lead to an exploding indicies problem, but we can fix that when

456 // we get to it.

457

458 //sortOrders := []qSortBy{}

459

460 return nil

461 }

462

463 func (q queryImpl) clone() queryImpl {

464 ret := *q

465 ret.filter = append([]queryFilter(nil), q.filter...)

466 ret.order = append([]queryOrder(nil), q.order...)

467 return &ret

468 }

469

470 func (q queryImpl) Ancestor(k datastore.Key) wrapper.DSQuery {

471 q = q.clone()

472 q.ancestor = k

473 if k == nil {

474 // SDK has an explicit nil-check

475 q.err = errors.New("datastore: nil query ancestor")

476 } else if !keyValid(q.ns, k, userKeyOnly) {

477 // technically the SDK implementation does a Weird Thing (tm) if both the

478 // stringID and intID are set on a key; it only serializes the s tringID in

479 // the proto. This means that if you set the Ancestor to an inva lid key,

480 // you'll never actually hear about it. Instead of doing that in sanity, we

481 // just swap to an error here.

482 q.err = datastore.ErrInvalidKey

483 }

484 return q

485 }

486

487 func (q *queryImpl) Filter(fStr string, val interface{}) wrapper.DSQuery {

488 q = q.clone()

489 f, err := parseFilter(fStr, val)

490 if err != nil {

491 q.err = err

492 return q

493 }

494 q.filter = append(q.filter, f)

495 return q

496 }

497

498 func (q *queryImpl) Order(field string) wrapper.DSQuery {

499 q = q.clone()

500 field = strings.TrimSpace(field)

501 o := queryOrder{field, qASC}

502 if strings.HasPrefix(field, "-") {

503 o.direction = qDEC

504 o.field = strings.TrimSpace(field[1:])

505 } else if strings.HasPrefix(field, "+") {

506 q.err = fmt.Errorf("datastore: invalid order: %q", field)

507 return q

508 }

509 if len(o.field) == 0 {

510 q.err = errors.New("datastore: empty order")

511 return q

512 }

513 q.order = append(q.order, o)

514 return q

515 }

516

517 func (q *queryImpl) KeysOnly() wrapper.DSQuery {

518 q = q.clone()

519 q.keysOnly = true

520 return q

521 }

522

523 func (q *queryImpl) Limit(limit int) wrapper.DSQuery {

524 q = q.clone()

525 if limit < math.MinInt32 \|\| limit > math.MaxInt32 {

526 q.err = errors.New("datastore: query limit overflow")

527 return q

528 }

529 q.limit = int32(limit)

530 return q

531 }

532

533 func (q *queryImpl) Offset(offset int) wrapper.DSQuery {

534 q = q.clone()

535 if offset < 0 {

536 q.err = errors.New("datastore: negative query offset")

537 return q

538 }

539 if offset > math.MaxInt32 {

540 q.err = errors.New("datastore: query offset overflow")

541 return q

542 }

543 q.offset = int32(offset)

544 return q

545 }

546

547 func (q *queryImpl) Start(c wrapper.DSCursor) wrapper.DSQuery {

548 q = q.clone()

549 curs := c.(queryCursor)

550 if !curs.Valid() {

551 q.err = errors.New("datastore: invalid cursor")

552 return q

553 }

554 q.start = curs

555 return q

556 }

557

558 func (q *queryImpl) End(c wrapper.DSCursor) wrapper.DSQuery {

559 q = q.clone()

560 curs := c.(queryCursor)

561 if !curs.Valid() {

562 q.err = errors.New("datastore: invalid cursor")

563 return q

564 }

565 q.end = curs

566 return q

567 }

OLD	NEW