1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
desc: Test changefeeds on a table
table_variable_name: tbl
tests:
# ==== regular tables
# - start feeds
- cd: all = tbl.changes()
# - note: no initial values from table changefeeds
# - inserts
- cd: tbl.insert([{'id':1}, {'id':2}])
ot: partial({'errors':0, 'inserted':2})
- cd: fetch(all, 2)
ot: bag([{'old_val':null, 'new_val':{'id':1}}, {'old_val':null, 'new_val':{'id':2}}])
# - updates
- cd: tbl.get(1).update({'version':1})
ot: partial({'errors':0, 'replaced':1})
- cd: fetch(all, 1)
ot: [{'old_val':{'id':1}, 'new_val':{'id':1, 'version':1}}]
# - deletions
- cd: tbl.get(1).delete()
ot: partial({'errors':0, 'deleted':1})
- cd: fetch(all, 1)
ot: [{'old_val':{'id':1, 'version':1}, 'new_val':null}]
# - pluck on values
- cd: pluck = tbl.changes().pluck({'new_val':['version']})
- cd: tbl.insert([{'id':5, 'version':5}])
ot: partial({'errors':0, 'inserted':1})
- cd: fetch(pluck, 1)
ot: [{'new_val':{'version':5}}]
# - order by
- cd: tbl.changes().order_by('id')
ot: err('ReqlQueryLogicError', "Cannot call a terminal (`reduce`, `count`, etc.) on an infinite stream (such as a changefeed).")
#
# ToDo: enable this when #4067 is done
#
# - js: orderedLimit = tbl.changes().limit(5).order_by(r.desc('id'))('new_val')('id')
# cd: orderedLimit = tbl.changes().limit(5).order_by(r.desc('id'))['new_val']['id']
# - js: tbl.range(100, 105).map(function (row) { return {'id':row} })
# py: tbl.range(100, 105).map({'id':r.row})
# rb: tbl.range(100, 105).map{|row| {'id':row}}
# - cd: fetch(orderedLimit)
# ot: [104, 103, 102, 101, 100]
# - changes overflow
- cd: overflow = tbl.changes()
runopts:
changefeed_queue_size: 100
# add enough entries to make sure we get the overflow error
- js: tbl.insert(r.range(200).map(function(x) { return({}); }))
py: tbl.insert(r.range(200).map(lambda x: {}))
rb: tbl.insert(r.range(200).map{|x| {}})
- cd: fetch(overflow, 90)
ot: partial([{'error': regex('Changefeed cache over array size limit, skipped \d+ elements.')}])
# ==== virtual tables
- def: vtbl = r.db('rethinkdb').table('_debug_scratch')
- cd: allVirtual = vtbl.changes()
# - inserts
- cd: vtbl.insert([{'id':1}, {'id':2}])
ot: partial({'errors':0, 'inserted':2})
- cd: fetch(allVirtual, 2)
ot: bag([{'old_val':null, 'new_val':{'id':1}}, {'old_val':null, 'new_val':{'id':2}}])
# - updates
- cd: vtbl.get(1).update({'version':1})
ot: partial({'errors':0, 'replaced':1})
- cd: fetch(allVirtual, 1)
ot: [{'old_val':{'id':1}, 'new_val':{'id':1, 'version':1}}]
# - deletions
- cd: vtbl.get(1).delete()
ot: partial({'errors':0, 'deleted':1})
- cd: fetch(allVirtual, 1)
ot: [{'old_val':{'id':1, 'version':1}, 'new_val':null}]
# - pluck on values
- cd: vpluck = vtbl.changes().pluck({'new_val':['version']})
- cd: vtbl.insert([{'id':5, 'version':5}])
ot: partial({'errors':0, 'inserted':1})
- cd: fetch(vpluck, 1)
ot: [{'new_val':{'version':5}}]
|
