1 import Base 2 import ../ByteMemTable.bend as ByteMemTable 3 import ../ByteWal.bend as ByteWal 4 import ./ByteBlockReader.bend as ByteBlockReader 5 import ./ByteRange.bend as ByteRange 6 import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes 7 8 # Carries half-open byte bounds and an optional exact byte prefix. 9 type Bounds is Type: 10 Bounds{lower: Maybe<&1, Bytes.Bytes>, upper: Maybe<&1, Bytes.Bytes>, 11 prefix: Maybe<&1, Bytes.Bytes>} 12 13 # Owns one validated SST reader and the unread part of its current block. 14 type Cursor is Type: 15 Active{inner: ByteBlockReader.EntryCursor, bounds: Bounds} 16 Ended{} 17 Closed{} 18 Failed{error: U32 & String} 19 20 # Represents a streamed record, filtered entry, end, or terminal error. 21 type Step is Type: 22 Item{entry: ByteMemTable.Entry, cursor: Cursor} 23 Skip{cursor: Cursor} 24 End{} 25 Error{error: U32 & String} 26 27 # Carries each bounded block validation transition before cursor publication. 28 type VerifyState is Type: 29 VerifyNeed{reader: ByteBlockReader.Reader, bounds: Bounds} 30 VerifyRead{result: Result<&1, &1, U32 & String, ByteBlockReader.Step>, 31 bounds: Bounds} 32 33 # Carries one block read or one current block during incremental scanning. 34 type ScanState is Type: 35 ScanNeed{reader: ByteBlockReader.Reader, bounds: Bounds} 36 ScanRead{result: Result<&1, &1, U32 & String, ByteBlockReader.Step>, 37 bounds: Bounds} 38 ScanBlock{reader: ByteBlockReader.Reader, 39 entries: List<&1, ByteMemTable.Entry>, bounds: Bounds} 40 41 def bounds.copy.values( 42 lower: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>, 43 upper: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>, 44 prefix: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes> 45 ) -> Bounds & Bounds: 46 match lower: 47 case (lower_original, lower_copy): 48 match upper: 49 case (upper_original, upper_copy): 50 match prefix: 51 case (prefix_original, prefix_copy): 52 (Bounds{lower_original, upper_original, prefix_original}, 53 Bounds{lower_copy, upper_copy, prefix_copy}) 54 55 def bounds.copy(bounds: Bounds) -> Bounds & Bounds: 56 match bounds: 57 case Bounds{lower, upper, prefix}: 58 bounds.copy.values(ByteMemTable.value.clone(lower), 59 ByteMemTable.value.clone(upper), ByteMemTable.value.clone(prefix)) 60 61 def scan.prefix(key: Bytes.Bytes, prefix: Maybe<&1, Bytes.Bytes>) -> Bool: 62 match prefix: 63 case None{}: True{} 64 case Some{bytes}: ByteRange.has_prefix(key, bytes) 65 66 def scan.entry.choose( 67 valid: Bool, 68 key: Bytes.Bytes, 69 value: Bytes.Bytes, 70 cursor: ByteBlockReader.EntryCursor, 71 bounds: Bounds & Bounds 72 ) -> IO(Step): 73 match valid: 74 case True{}: 75 match bounds: 76 case (item_bounds, _): 77 IO.pure(Step, Item{ByteMemTable.Entry{key, Some{value}}, 78 Active{cursor, item_bounds}}) 79 case False{}: 80 match bounds: 81 case (_, skip_bounds): IO.pure(Step, Skip{Active{cursor, skip_bounds}}) 82 83 def scan.entry.checked( 84 key: Bytes.Bytes, 85 value: Maybe<&1, Bytes.Bytes>, 86 cursor: ByteBlockReader.EntryCursor, 87 bounds: Bounds & Bounds, 88 key_copies: Bytes.Bytes & Bytes.Bytes 89 ) -> IO(Step): 90 match value: 91 case Some{bytes}: 92 match bounds: 93 case (checked, retained): 94 match checked: 95 case Bounds{lower, upper, prefix}: 96 match key_copies: 97 case (range_key, prefix_key): 98 scan.entry.choose( 99 ByteRange.in_range(range_key, lower, upper) && 100 scan.prefix(prefix_key, prefix), 101 key, bytes, cursor, bounds.copy(retained)) 102 case None{}: 103 match bounds: 104 case (_, retained): IO.pure(Step, Skip{Active{cursor, retained}}) 105 106 def scan.entry.upper.closed( 107 closed: Result<&1, &1, U32 & String, Unit> 108 ) -> Step: 109 match closed: 110 case Done{Unit{}}: End{} 111 case Fail{error}: Error{error} 112 113 def scan.entry.upper.result( 114 reached: Bool, 115 key: Bytes.Bytes, 116 value: Maybe<&1, Bytes.Bytes>, 117 cursor: ByteBlockReader.EntryCursor, 118 lower: Maybe<&1, Bytes.Bytes>, 119 prefix: Maybe<&1, Bytes.Bytes>, 120 retained: Bounds, 121 upper_checked: Maybe<&1, Bytes.Bytes>, 122 key_copies: Bytes.Bytes & Bytes.Bytes 123 ) -> IO(Step): 124 match reached: 125 case True{}: 126 do IO: 127 closed : Result<&1, &1, U32 & String, Unit> <- 128 ByteBlockReader.entry.close(cursor) 129 return scan.entry.upper.closed(closed) 130 case False{}: 131 scan.entry.checked(key, value, cursor, 132 (Bounds{lower, upper_checked, prefix}, retained), key_copies) 133 134 def scan.entry.upper.copies( 135 key: Bytes.Bytes, 136 value: Maybe<&1, Bytes.Bytes>, 137 cursor: ByteBlockReader.EntryCursor, 138 lower: Maybe<&1, Bytes.Bytes>, 139 prefix: Maybe<&1, Bytes.Bytes>, 140 retained: Bounds, 141 upper: Maybe<&1, Bytes.Bytes> & Maybe<&1, Bytes.Bytes>, 142 key_copies: Bytes.Bytes & Bytes.Bytes 143 ) -> IO(Step): 144 match upper: 145 case (upper_test, upper_checked): 146 match key_copies: 147 case (upper_key, filter_key): 148 scan.entry.upper.result( 149 ByteRange.at_or_after_upper(upper_key, upper_test), 150 key, value, cursor, lower, prefix, retained, 151 upper_checked, 152 ByteMemTable.bytes.clone(filter_key)) 153 154 def scan.entry.upper.fields( 155 key: Bytes.Bytes, 156 value: Maybe<&1, Bytes.Bytes>, 157 cursor: ByteBlockReader.EntryCursor, 158 lower: Maybe<&1, Bytes.Bytes>, 159 upper: Maybe<&1, Bytes.Bytes>, 160 prefix: Maybe<&1, Bytes.Bytes>, 161 retained: Bounds, 162 key_copies: Bytes.Bytes & Bytes.Bytes 163 ) -> IO(Step): 164 scan.entry.upper.copies(key, value, cursor, lower, prefix, retained, 165 ByteMemTable.value.clone(upper), key_copies) 166 167 def scan.entry.upper.checked( 168 key: Bytes.Bytes, 169 value: Maybe<&1, Bytes.Bytes>, 170 cursor: ByteBlockReader.EntryCursor, 171 bounds: Bounds & Bounds, 172 key_copies: Bytes.Bytes & Bytes.Bytes 173 ) -> IO(Step): 174 match bounds: 175 case (checked, retained): 176 match checked: 177 case Bounds{lower, upper, prefix}: 178 scan.entry.upper.fields(key, value, cursor, lower, upper, prefix, 179 retained, key_copies) 180 181 def scan.entry.key.copied( 182 value: Maybe<&1, Bytes.Bytes>, 183 cursor: ByteBlockReader.EntryCursor, 184 bounds: Bounds & Bounds, 185 pair: Bytes.Bytes & Bytes.Bytes 186 ) -> IO(Step): 187 match pair: 188 case (key, check_key): 189 scan.entry.upper.checked(key, value, cursor, bounds, 190 ByteMemTable.bytes.clone(check_key)) 191 192 def scan.entry( 193 entry: ByteMemTable.Entry, 194 cursor: ByteBlockReader.EntryCursor, 195 bounds: Bounds 196 ) -> IO(Step): 197 match entry: 198 case ByteMemTable.Entry{key, value}: 199 scan.entry.key.copied(value, cursor, bounds.copy(bounds), 200 ByteMemTable.bytes.clone(key)) 201 202 def scan.exhausted(state: ScanState) -> IO(Step): 203 match state: 204 case ScanNeed{reader, _}: 205 do IO: 206 _closed : Result<&1, &1, U32 & String, Unit> <- ByteBlockReader.close(reader) 207 return Error{(7, "SST cursor exceeded its transition budget")} 208 case ScanRead{result, _}: 209 match result: 210 case Done{ByteBlockReader.Block{reader, _}}: 211 do IO: 212 _closed : Result<&1, &1, U32 & String, Unit> <- 213 ByteBlockReader.close(reader) 214 return Error{(7, "SST cursor exceeded its transition budget")} 215 case _: 216 IO.pure(Step, Error{(7, "SST cursor exceeded its transition budget")}) 217 case ScanBlock{reader, _, _}: 218 do IO: 219 _closed : Result<&1, &1, U32 & String, Unit> <- ByteBlockReader.close(reader) 220 return Error{(7, "SST cursor exceeded its transition budget")} 221 222 def scan.loop(fuel: Nat, state: ScanState) -> IO(Step): 223 match fuel: 224 case 0n: scan.exhausted(state) 225 case 1n+rest: 226 match state: 227 case ScanNeed{reader, bounds}: 228 do IO: 229 result : Result<&1, &1, U32 & String, ByteBlockReader.Step> <- 230 ByteBlockReader.next(reader) 231 scan.loop(rest, ScanRead{result, bounds}) 232 case ScanRead{result, bounds}: 233 match result: 234 case Fail{error}: 235 IO.pure(Step, Error{error}) 236 case Done{ByteBlockReader.End{}}: 237 IO.pure(Step, End{}) 238 case Done{ByteBlockReader.Block{reader, entries}}: 239 scan.loop(rest, ScanBlock{reader, entries, bounds}) 240 case ScanBlock{reader, entries, bounds}: 241 match entries: 242 case Nil{}: 243 scan.loop(rest, ScanNeed{reader, bounds}) 244 case Con{entry, tail}: 245 scan.entry(entry, ByteBlockReader.EntryReady{reader, tail}, bounds) 246 247 def verify.exhausted(state: VerifyState) -> IO(Result<&1, &1, U32 & String, Cursor>): 248 match state: 249 case VerifyNeed{reader, _}: 250 do IO>: 251 _closed : Result<&1, &1, U32 & String, Unit> <- ByteBlockReader.close(reader) 252 return Fail{(7, "SST validation exceeded its transition budget")} 253 case VerifyRead{result, _}: 254 match result: 255 case Done{ByteBlockReader.Block{reader, _}}: 256 do IO>: 257 _closed : Result<&1, &1, U32 & String, Unit> <- 258 ByteBlockReader.close(reader) 259 return Fail{(7, "SST validation exceeded its transition budget")} 260 case _: 261 IO.pure(Result<&1, &1, U32 & String, Cursor>, 262 Fail{(7, "SST validation exceeded its transition budget")}) 263 264 def verify.complete.routed.result( 265 bounds: Bounds, 266 result: Result<&1, &1, U32 & String, ByteBlockReader.Reader> 267 ) -> IO(Result<&1, &1, U32 & String, Cursor>): 268 match result: 269 case Fail{error}: 270 IO.pure(Result<&1, &1, U32 & String, Cursor>, Fail{error}) 271 case Done{reader}: 272 IO.pure(Result<&1, &1, U32 & String, Cursor>, 273 Done{Active{ByteBlockReader.entry.open(reader), bounds}}) 274 275 def verify.complete.routed( 276 bounds: Bounds, 277 result: IO(Result<&1, &1, U32 & String, ByteBlockReader.Reader>) 278 ) -> IO(Result<&1, &1, U32 & String, Cursor>): 279 do IO>: 280 routed : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <- result 281 verify.complete.routed.result(bounds, routed) 282 283 def verify.complete.copied( 284 pair: Bounds & Bounds, 285 reader: ByteBlockReader.Reader 286 ) -> IO(Result<&1, &1, U32 & String, Cursor>): 287 match pair: 288 case (retained, seek_bounds): 289 match seek_bounds: 290 case Bounds{lower, _, _}: 291 match lower: 292 case None{}: 293 IO.pure(Result<&1, &1, U32 & String, Cursor>, 294 Done{Active{ByteBlockReader.entry.open(reader), retained}}) 295 case Some{key}: 296 verify.complete.routed(retained, 297 ByteBlockReader.seek_key_after_validation(reader, key)) 298 299 def verify.complete( 300 bounds: Bounds, 301 result: Result<&1, &1, U32 & String, ByteBlockReader.Reader> 302 ) -> IO(Result<&1, &1, U32 & String, Cursor>): 303 match result: 304 case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Cursor>, Fail{error}) 305 case Done{reader}: 306 verify.complete.copied(bounds.copy(bounds), reader) 307 308 def verify.loop( 309 fuel: Nat, 310 path: String, 311 state: VerifyState 312 ) -> IO(Result<&1, &1, U32 & String, Cursor>): 313 match fuel: 314 case 0n: verify.exhausted(state) 315 case 1n+rest: 316 match state: 317 case VerifyNeed{reader, bounds}: 318 do IO>: 319 result : Result<&1, &1, U32 & String, ByteBlockReader.Step> <- 320 ByteBlockReader.next(reader) 321 verify.loop(rest, path, VerifyRead{result, bounds}) 322 case VerifyRead{result, bounds}: 323 match result: 324 case Fail{error}: 325 IO.pure(Result<&1, &1, U32 & String, Cursor>, Fail{error}) 326 case Done{ByteBlockReader.End{}}: 327 do IO>: 328 reopened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <- 329 ByteBlockReader.open(path) 330 verify.complete(bounds, reopened) 331 case Done{ByteBlockReader.Block{reader, _}}: 332 verify.loop(rest, path, VerifyNeed{reader, bounds}) 333 334 def open.budget( 335 path: String, 336 bounds: Bounds, 337 pair: Nat & ByteBlockReader.Reader 338 ) -> IO(Result<&1, &1, U32 & String, Cursor>): 339 match pair: 340 case (fuel, reader): 341 verify.loop(fuel, path, VerifyNeed{reader, bounds}) 342 343 def open.result( 344 path: String, 345 bounds: Bounds, 346 opened: Result<&1, &1, U32 & String, ByteBlockReader.Reader> 347 ) -> IO(Result<&1, &1, U32 & String, Cursor>): 348 match opened: 349 case Fail{error}: IO.pure(Result<&1, &1, U32 & String, Cursor>, Fail{error}) 350 case Done{reader}: 351 open.budget(path, bounds, ByteBlockReader.transition_budget(reader)) 352 353 def open.trusted.result( 354 bounds: Bounds, 355 opened: Result<&1, &1, U32 & String, ByteBlockReader.Reader> 356 ) -> IO(Result<&1, &1, U32 & String, Cursor>): 357 match opened: 358 case Fail{error}: 359 IO.pure(Result<&1, &1, U32 & String, Cursor>, Fail{error}) 360 case Done{reader}: 361 verify.complete(bounds, Done{reader}) 362 363 # Reopens a previously validated generation without rescanning every block. 364 def open.trusted( 365 +path: String, 366 bounds: Bounds 367 ) -> IO(Result<&1, &1, U32 & String, Cursor>): 368 do IO>: 369 opened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <- 370 ByteBlockReader.open(path) 371 open.trusted.result(bounds, opened) 372 373 def measure.key.opened( 374 key: Bytes.Bytes, 375 opened: Result<&1, &1, U32 & String, ByteBlockReader.Reader> 376 ) -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>): 377 match opened: 378 case Fail{error}: 379 IO.pure(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>, 380 Fail{error}) 381 case Done{reader}: ByteBlockReader.measure_key(reader, key) 382 383 # Handles measure key. 384 def measure_key(path: String, key: Bytes.Bytes) 385 -> IO(Result<&1, &1, U32 & String, ByteWal.ValueMetadata>): 386 do IO>: 387 opened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <- 388 ByteBlockReader.open(path) 389 measure.key.opened(key, opened) 390 391 # Validates the complete SST before opening a second bounded scan pass. 392 def open( 393 +path: String, 394 bounds: Bounds 395 ) -> 396 IO(Result<&1, &1, U32 & String, Cursor>): 397 do IO>: 398 opened : Result<&1, &1, U32 & String, ByteBlockReader.Reader> <- 399 ByteBlockReader.open(path) 400 open.result(path, bounds, opened) 401 402 def scan.start(pair: Nat & ByteBlockReader.Reader, bounds: Bounds) -> IO(Step): 403 match pair: 404 case (fuel, reader): scan.loop(fuel, ScanNeed{reader, bounds}) 405 406 # Returns the next bounded record, range skip, end, or terminal read error. 407 def next(cursor: Cursor) -> IO(Step): 408 match cursor: 409 case Active{inner, bounds}: 410 match inner: 411 case ByteBlockReader.EntryReady{reader, entries}: 412 match entries: 413 case Con{entry, tail}: 414 scan.entry(entry, ByteBlockReader.EntryReady{reader, tail}, bounds) 415 case Nil{}: 416 scan.start(ByteBlockReader.transition_budget(reader), bounds) 417 case ByteBlockReader.EntryEnded{}: 418 IO.pure(Step, End{}) 419 case ByteBlockReader.EntryClosed{}: 420 IO.pure(Step, End{}) 421 case ByteBlockReader.EntryFailed{error}: 422 IO.pure(Step, Error{error}) 423 case Ended{}: IO.pure(Step, End{}) 424 case Closed{}: IO.pure(Step, End{}) 425 case Failed{error}: IO.pure(Step, Error{error}) 426 427 # Advances the remaining scan to an inclusive lower byte key. 428 def seek(cursor: Cursor, lower: Maybe<&1, Bytes.Bytes>) -> Cursor: 429 match cursor: 430 case Active{inner, bounds}: 431 match bounds: 432 case Bounds{_, upper, prefix}: 433 Active{inner, Bounds{lower, upper, prefix}} 434 case Ended{}: Ended{} 435 case Closed{}: Closed{} 436 case Failed{error}: Failed{error} 437 438 # Releases the active file reader, preserving terminal cursors. 439 def close(cursor: Cursor) -> IO(Result<&1, &1, U32 & String, Unit>): 440 match cursor: 441 case Active{inner, _}: ByteBlockReader.entry.close(inner) 442 case _: 443 IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})