1	import Base
2	import ../ByteMemTable.bend as ByteMemTable
3	import ../Keys.bend as Keys
4	import ./ByteRange.bend as ByteRange
5	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
6	
7	# Carries optional half-open range bounds and an optional byte prefix.
8	type Bounds is Type:
9	  Bounds{lower: Maybe<&1, Bytes.Bytes>, upper: Maybe<&1, Bytes.Bytes>, prefix: Maybe<&1, Bytes.Bytes>}
10	
11	# Owns the unread suffix of one ordered byte-entry source.
12	type Cursor is Type:
13	  RangeReady{entries: List<&1, ByteMemTable.Entry>}
14	  RangeEnded{}
15	  RangeClosed{}
16	
17	# Returns an item, a filtered skip, or terminal end with retained bounds.
18	type Step is Type:
19	  RangeItem{entry: ByteMemTable.Entry, cursor: Cursor, bounds: Bounds}
20	  RangeSkip{cursor: Cursor, bounds: Bounds}
21	  RangeEnd{cursor: Cursor, bounds: Bounds}
22	
23	# Returns the first entry at or above a lower bound.
24	type SeekStep is Type:
25	  SeekSkip{entries: List<&1, ByteMemTable.Entry>, lower: Bytes.Bytes}
26	  SeekFound{entries: List<&1, ByteMemTable.Entry>}
27	
28	def seek.order(
29	  key: Bytes.Bytes,
30	  value: Maybe<&1, Bytes.Bytes>,
31	  tail: List<&1, ByteMemTable.Entry>,
32	  lower: Bytes.Bytes,
33	  comparison: Bytes.Bytes & Bytes.Bytes & Cmp
34	) -> SeekStep:
35	  match comparison:
36	    case (_, _, LT{}): SeekSkip{tail, lower}
37	    case _ : SeekFound{Con{ByteMemTable.Entry{key, value}, tail}}
38	
39	def seek.compared(
40	  key: Bytes.Bytes,
41	  key_check: Bytes.Bytes,
42	  value: Maybe<&1, Bytes.Bytes>,
43	  tail: List<&1, ByteMemTable.Entry>,
44	  pair: Bytes.Bytes & Bytes.Bytes
45	) -> SeekStep:
46	  match pair:
47	    case (lower, lower_check):
48	      seek.order(key, value, tail, lower, Keys.Bytes.cmp.with_inputs(key_check, lower_check))
49	
50	def seek.key.copies(
51	  key: Bytes.Bytes,
52	  key_check: Bytes.Bytes,
53	  value: Maybe<&1, Bytes.Bytes>,
54	  tail: List<&1, ByteMemTable.Entry>,
55	  lower: Bytes.Bytes
56	) -> SeekStep:
57	  seek.compared(key, key_check, value, tail, ByteMemTable.bytes.clone(lower))
58	
59	def seek.entry.cloned(
60	  value: Maybe<&1, Bytes.Bytes>,
61	  tail: List<&1, ByteMemTable.Entry>,
62	  lower: Bytes.Bytes,
63	  pair: Bytes.Bytes & Bytes.Bytes
64	) -> SeekStep:
65	  match pair:
66	    case (original_key, check_key):
67	      seek.key.copies(original_key, check_key, value, tail, lower)
68	
69	def seek.entry(
70	  entry: ByteMemTable.Entry,
71	  tail: List<&1, ByteMemTable.Entry>,
72	  lower: Bytes.Bytes
73	) -> SeekStep:
74	  match entry:
75	    case ByteMemTable.Entry{key, value}:
76	      seek.entry.cloned(value, tail, lower, ByteMemTable.bytes.clone(key))
77	
78	def seek.step(
79	  entry: ByteMemTable.Entry,
80	  tail: List<&1, ByteMemTable.Entry>,
81	  lower: Maybe<&1, Bytes.Bytes>
82	) -> SeekStep:
83	  match lower:
84	    case None{}: SeekFound{Con{entry, tail}}
85	    case Some{bound}: seek.entry(entry, tail, bound)
86	
87	def seek.first(
88	  entries: List<&1, ByteMemTable.Entry>,
89	  lower: Maybe<&1, Bytes.Bytes>
90	) -> SeekStep:
91	  match entries:
92	    case Nil{}: SeekFound{Nil{}}
93	    case Con{entry, tail}: seek.step(entry, tail, lower)
94	
95	def seek.go(fuel: Nat, result: SeekStep) -> List<&1, ByteMemTable.Entry>:
96	  match fuel:
97	    case 0n:
98	      match result:
99	        case SeekSkip{entries, _}: entries
100	        case SeekFound{found}: found
101	    case 1n+rest:
102	      match result:
103	        case SeekSkip{entries, lower}:
104	          seek.go(rest, seek.first(entries, Some{lower}))
105	        case SeekFound{found}: found
106	
107	def seek.counted(
108	  entries: List<&1, ByteMemTable.Entry> & Nat,
109	  lower: Maybe<&1, Bytes.Bytes>
110	) -> List<&1, ByteMemTable.Entry>:
111	  match entries:
112	    case (source, count): seek.go(Nat.add(count, 1n), seek.first(source, lower))
113	
114	# Drops ordered entries below the inclusive lower bound.
115	def seek(
116	  entries: List<&1, ByteMemTable.Entry>,
117	  lower: Maybe<&1, Bytes.Bytes>
118	) -> List<&1, ByteMemTable.Entry>:
119	  seek.counted(ByteMemTable.count.keep(entries), lower)
120	
121	# Moves an active source forward to the first key at or above the lower bound.
122	def seek.cursor(cursor: Cursor, lower: Maybe<&1, Bytes.Bytes>) -> Cursor:
123	  match cursor:
124	    case RangeReady{entries}: RangeReady{seek(entries, lower)}
125	    case RangeEnded{}: RangeEnded{}
126	    case RangeClosed{}: RangeClosed{}
127	
128	def bounds.clone.parts(
129	  lower: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>,
130	  upper: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>,
131	  prefix: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>
132	) -> Bounds & Bounds:
133	  match lower:
134	    case (lower_original, lower_copy):
135	      match upper:
136	        case (upper_original, upper_copy):
137	          match prefix:
138	            case (prefix_original, prefix_copy):
139	              (Bounds{lower_original, upper_original, prefix_original},
140	                Bounds{lower_copy, upper_copy, prefix_copy})
141	
142	def bounds.clone(bounds: Bounds) -> Bounds & Bounds:
143	  match bounds:
144	    case Bounds{lower, upper, prefix}:
145	      bounds.clone.parts(ByteMemTable.value.clone(lower),
146	        ByteMemTable.value.clone(upper), ByteMemTable.value.clone(prefix))
147	
148	def matches.prefix(key: Bytes.Bytes, prefix: Maybe<&1, Bytes.Bytes>) -> Bool:
149	  match prefix:
150	    case None{}: True{}
151	    case Some{value}: ByteRange.has_prefix(key, value)
152	
153	def matches.key(pair: Bytes.Bytes & Bytes.Bytes, bounds: Bounds) -> Bool:
154	  match pair:
155	    case (range_key, prefix_key):
156	      match bounds:
157	        case Bounds{lower, upper, prefix}:
158	          Bool.and(ByteRange.in_range(range_key, lower, upper),
159	            matches.prefix(prefix_key, prefix))
160	
161	def filtered.result(
162	  valid: Bool,
163	  key: Bytes.Bytes,
164	  value: Bytes.Bytes,
165	  tail: List<&1, ByteMemTable.Entry>,
166	  bounds: Bounds
167	) -> Step:
168	  match valid:
169	    case True{}:
170	      RangeItem{ByteMemTable.Entry{key, Some{value}}, RangeReady{tail}, bounds}
171	    case False{}:
172	      RangeSkip{RangeReady{tail}, bounds}
173	
174	def key.copies.pair(
175	  original: Bytes.Bytes,
176	  copies: Bytes.Bytes & Bytes.Bytes
177	) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
178	  match copies:
179	    case (range_key, prefix_key):
180	      (original, range_key, prefix_key)
181	
182	def key.copies.first(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
183	  match pair:
184	    case (original, check_key):
185	      key.copies.pair(original, ByteMemTable.bytes.clone(check_key))
186	
187	def key.copies(key: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
188	  key.copies.first(ByteMemTable.bytes.clone(key))
189	
190	def filtered.bounds(
191	  value: Bytes.Bytes,
192	  tail: List<&1, ByteMemTable.Entry>,
193	  bounds: Bounds & Bounds,
194	  keys: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes
195	) -> Step:
196	  match bounds:
197	    case (checked, retained):
198	      match keys:
199	        case (original_key, range_key, prefix_key):
200	          filtered.result(matches.key((range_key, prefix_key), checked),
201	            original_key, value, tail, retained)
202	
203	def filtered.key(
204	  keys: Bytes.Bytes & Bytes.Bytes,
205	  value: Bytes.Bytes,
206	  tail: List<&1, ByteMemTable.Entry>,
207	  bounds: Bounds & Bounds
208	) -> Step:
209	  match keys:
210	    case (key, check_key):
211	      filtered.bounds(value, tail, bounds, key.copies(check_key))
212	
213	def filtered.value(
214	  key: Bytes.Bytes,
215	  value: Maybe<&1, Bytes.Bytes>,
216	  tail: List<&1, ByteMemTable.Entry>,
217	  bounds: Bounds & Bounds
218	) -> Step:
219	  match value:
220	    case None{}:
221	      match bounds:
222	        case (_, retained): RangeSkip{RangeReady{tail}, retained}
223	    case Some{bytes}:
224	      filtered.key(ByteMemTable.bytes.clone(key), bytes, tail, bounds)
225	
226	# Opens an ordered source for range and prefix filtering.
227	def open(entries: List<&1, ByteMemTable.Entry>) -> Cursor:
228	  RangeReady{entries}
229	
230	def next.pair.checked(
231	  reached_upper: Bool,
232	  key: Bytes.Bytes,
233	  value: Maybe<&1, Bytes.Bytes>,
234	  tail: List<&1, ByteMemTable.Entry>,
235	  bounds: Bounds & Bounds
236	) -> Step:
237	  match reached_upper:
238	    case True{}:
239	      match bounds:
240	        case (_, retained): RangeEnd{RangeEnded{}, retained}
241	    case False{}:
242	      filtered.value(key, value, tail, bounds)
243	
244	def next.upper.test.bytes(
245	  keys: Bytes.Bytes & Bytes.Bytes,
246	  upper: Maybe<&1, Bytes.Bytes>
247	) -> Bool:
248	  match keys:
249	    case (_, check_key): ByteRange.at_or_after_upper(check_key, upper)
250	
251	def next.upper.test(
252	  key: Bytes.Bytes,
253	  bounds: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>
254	) -> Bool:
255	  match bounds:
256	    case (upper, _):
257	      next.upper.test.bytes(ByteMemTable.bytes.clone(key), upper)
258	
259	def next.pair.upper(key: Bytes.Bytes, bounds: Bounds) -> Bool:
260	  match bounds:
261	    case Bounds{_, upper, _}:
262	      next.upper.test(key, ByteMemTable.value.clone(upper))
263	
264	def next.pair.bounds.copied(
265	  key: Bytes.Bytes,
266	  check_key: Bytes.Bytes,
267	  value: Maybe<&1, Bytes.Bytes>,
268	  tail: List<&1, ByteMemTable.Entry>,
269	  bounds: Bounds & Bounds,
270	  retained: Bounds
271	) -> Step:
272	  match bounds:
273	    case (upper_bounds, filter_bounds):
274	      next.pair.checked(next.pair.upper(check_key, upper_bounds), key,
275	        value, tail, (filter_bounds, retained))
276	
277	def next.pair.bounds(
278	  key: Bytes.Bytes,
279	  check_key: Bytes.Bytes,
280	  value: Maybe<&1, Bytes.Bytes>,
281	  tail: List<&1, ByteMemTable.Entry>,
282	  bounds: Bounds & Bounds
283	) -> Step:
284	  match bounds:
285	    case (checked, retained):
286	      next.pair.bounds.copied(key, check_key, value, tail,
287	        bounds.clone(checked), retained)
288	
289	def next.pair.cloned(
290	  keys: Bytes.Bytes & Bytes.Bytes,
291	  value: Maybe<&1, Bytes.Bytes>,
292	  tail: List<&1, ByteMemTable.Entry>,
293	  bounds: Bounds & Bounds
294	) -> Step:
295	  match keys:
296	    case (key, check_key):
297	      next.pair.bounds(key, check_key, value, tail, bounds)
298	
299	def next.pair(
300	  entry: ByteMemTable.Entry,
301	  tail: List<&1, ByteMemTable.Entry>,
302	  bounds: Bounds & Bounds
303	) -> Step:
304	  match entry:
305	    case ByteMemTable.Entry{key, value}:
306	      next.pair.cloned(ByteMemTable.bytes.clone(key), value, tail, bounds)
307	
308	def next.ready(
309	  entries: List<&1, ByteMemTable.Entry>,
310	  bounds: Bounds & Bounds
311	) -> Step:
312	  match entries:
313	    case Nil{}:
314	      match bounds:
315	        case (_, retained): RangeEnd{RangeEnded{}, retained}
316	    case Con{entry, tail}:
317	      next.pair(entry, tail, bounds)
318	
319	# Returns one match, a skip event, or end while returning the bounds.
320	def next(cursor: Cursor, bounds: Bounds & Bounds) -> Step:
321	  match cursor:
322	    case RangeReady{entries}:
323	      next.ready(entries, bounds)
324	    case RangeEnded{}:
325	      match bounds:
326	        case (_, retained): RangeEnd{RangeEnded{}, retained}
327	    case RangeClosed{}:
328	      match bounds:
329	        case (_, retained): RangeEnd{RangeClosed{}, retained}
330	
331	# Closes an active byte range cursor.
332	def close(cursor: Cursor) -> Cursor:
333	  match cursor:
334	    case RangeReady{_}: RangeClosed{}
335	    case RangeEnded{}: RangeEnded{}
336	    case RangeClosed{}: RangeClosed{}