def table.walk.go

mylsm-lsm-store@0.6.0.1 · src/transfer/MigrateIo.bend:242

import Base
import mylsm-lsm-store@0.6.0.1/src/Sstable.bend as Sstable
import mylsm-lsm-store@0.6.0.1/src/SstReadPolicy.bend as SstReadPolicy
import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as QueuePolicy

242def table.walk.go(fuel: Nat, state: LegacyTableWalk) -> IO(TableValidation):
243  match fuel:
244    case 0n:
245      match state:
246        case LegacyLevels{lock, _, release_path, _, _, _, _, records}:
247          table.finished(lock, release_path, Fail{ResourceLimit{}})
248        case LegacyPath{lock, _, release_path, _, _, _, _, _, _, records}:
249          table.finished(lock, release_path, Fail{ResourceLimit{}})
250        case LegacyNames{lock, _, release_path, _, _, _, _, _, _, records}:
251          table.finished(lock, release_path, Fail{ResourceLimit{}})
252        case LegacySize{lock, _, release_path, _, _, _, _, _, _, _, _, records}:
253          table.finished(lock, release_path, Fail{ResourceLimit{}})
254        case LegacyBudget{lock, _, release_path, _, _, _, _, _, _, _, records}:
255          table.finished(lock, release_path, Fail{ResourceLimit{}})
256        case LegacyRead{lock, _, release_path, _, _, _, _, _, _, _, records}:
257          table.finished(lock, release_path, Fail{ResourceLimit{}})
258    case 1n+rest:
259      match state:
260        case LegacyLevels{lock, read_path, release_path, levels, index,
261            total_bytes, entries, records}:
262          match levels:
263            case Nil{}:
264              table.finished.records(lock, release_path,
265                Done{(entries, total_bytes)}, records)
266            case Con{names, tail}:
267              table.walk.go(rest, table.walk.level.path(
268                QueuePolicy.nat.clone.pair(index), lock, read_path,
269                release_path, names, tail, total_bytes, entries, records))
270        case LegacyPath{lock, read_path, release_path, ldir, names, levels,
271            index, total_bytes, entries, records}:
272          match ldir:
273            case None{}:
274              table.finished.records(lock, release_path,
275                Fail{InvalidLegacyManifest{}}, records)
276            case Some{dir}:
277              table.walk.go(rest, LegacyNames{lock, read_path, release_path,
278                levels, index, dir, names, total_bytes, entries, records})
279        case LegacyNames{lock, read_path, release_path, levels, index, dir,
280            names, total_bytes, entries, records}:
281          match names:
282            case Nil{}:
283              table.walk.go(rest, LegacyLevels{lock, read_path, release_path,
284                levels, Nat.add(index, 1n), total_bytes, entries, records})
285            case Con{name, tail}:
286              do IO<TableValidation>:
287                next : LegacyTableWalk <- table.walk.size(
288                  (path.clone.pair(read_path),
289                  (path.clone.pair(name), path.clone.pair(dir))),
290                  lock, release_path, levels, index, tail,
291                  total_bytes, entries, records)
292                table.walk.go(rest, next)
293        case LegacySize{lock, read_path, release_path, levels, index, dir,
294            name, names, total_bytes, entries, result, records}:
295          match result:
296            case Fail{(code, message)}:
297              table.finished.records(lock, release_path,
298                Fail{HostError{code, message}}, records)
299            case Done{size}:
300              table.walk.go(rest, LegacyBudget{lock, read_path, release_path,
301                levels, index, dir, name, names, entries,
302                SstReadPolicy.total_decision(total_bytes, size, 67108864n),
303                records})
304        case LegacyBudget{lock, read_path, release_path, levels, index, dir,
305            name, names, entries, decision, records}:
306          match decision:
307            case SstReadPolicy.TotalRejected{}:
308              table.finished.records(lock, release_path,
309                Fail{ResourceLimit{}}, records)
310            case SstReadPolicy.TotalAdmitted{next_total}:
311              do IO<TableValidation>:
312                next : LegacyTableWalk <- table.walk.decode(
313                  (path.clone.pair(read_path), path.clone.pair(dir)), lock,
314                  release_path, levels, index, name, names, next_total,
315                  entries, records)
316                table.walk.go(rest, next)
317        case LegacyRead{lock, read_path, release_path, levels, index, dir,
318            names, total_bytes, entries, result, records}:
319          match result:
320            case Fail{(code, _)}:
321              table.finished.records(lock, release_path,
322                Fail{table.failure(code)}, records)
323            case Done{Sstable.Tbl{table_entries, _, _, _, _, _, _, _}}:
324              table.walk.go(rest, table.read.next(
325                table.entries.count.keep(table_entries), lock, read_path,
326                release_path, levels, index, dir, names, total_bytes,
327                entries, records))