diff --git a/README.md b/README.md index 38f52e4..1c49fc4 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,14 @@ A Python package for versioning and monitoring dataset changes throughout the ma `dstrack` helps data scientists and ML engineers track how datasets evolve over time, catching schema drift, distribution shifts, and unexpected mutations before they silently break pipelines or degrade model performance. +## What dstrack is (and isn't) + +`dstrack` records *semantics* about a dataset - its schema, a content fingerprint, and per-column statistics - not the dataset itself. +It is not a data version control system: unlike DVC, LakeFS, or Git LFS, it never copies your files into a store or pushes them to a +remote, and it cannot check an old version of `data.csv` back out for you. Your data stays wherever it already lives, and `dstrack` keeps a small, +human-readable audit trail beside your code so you can see how that data changed and when. If you need to store and retrieve the bytes, reach for +DVC and use `dstrack` alongside it. + ## Installation ```bash diff --git a/docs/API/paths.md b/docs/API/paths.md new file mode 100644 index 0000000..4171c3c --- /dev/null +++ b/docs/API/paths.md @@ -0,0 +1,5 @@ +--- +icon: lucide/route +--- + +::: dstrack.paths diff --git a/docs/API/store.md b/docs/API/store.md new file mode 100644 index 0000000..7559895 --- /dev/null +++ b/docs/API/store.md @@ -0,0 +1,5 @@ +--- +icon: lucide/file-stack +--- + +::: dstrack.store diff --git a/docs/assets/dstrack-favicon.png b/docs/assets/dstrack-favicon.png index f896352..c430ff4 100644 Binary files a/docs/assets/dstrack-favicon.png and b/docs/assets/dstrack-favicon.png differ diff --git a/docs/assets/dstrack-logo.png b/docs/assets/dstrack-logo.png index 9465ee7..ae0d18b 100644 Binary files a/docs/assets/dstrack-logo.png and b/docs/assets/dstrack-logo.png differ diff --git a/docs/assets/dstrack-logo.svg b/docs/assets/dstrack-logo.svg index 0696f63..dafb832 100644 --- a/docs/assets/dstrack-logo.svg +++ b/docs/assets/dstrack-logo.svg @@ -1,4 +1,4 @@ - + dstrack - logo - + diff --git a/docs/decisions/0003-local-snapshot-store-layout.md b/docs/decisions/0003-local-snapshot-store-layout.md index ad785d6..7d85218 100644 --- a/docs/decisions/0003-local-snapshot-store-layout.md +++ b/docs/decisions/0003-local-snapshot-store-layout.md @@ -44,9 +44,9 @@ number of files inside `.dstrack/` are the exception, see ``` .dstrack/ -├── .gitignore # excludes cache/ -├── cache/ -│ └── index.db # gitignored: disposable search accelerator +├── .gitignore # excludes .cache/ +├── .cache/ +│ └── index.db # gitignored: disposable query accelerator └── datasets/ └── / # dataset_id: UUID4, minted at first snapshot, permanent ├── HEAD # snapshot_id of the latest snapshot @@ -73,14 +73,14 @@ satisfies both. ADR-0001; `dataset_id` groups all of a dataset's snapshots together. ### History and search: what's derived, what's not -Two different jobs need two different mechanisms: - -- **History of one dataset** (`dstrack log `) only ever reads that - dataset's own `snapshots/`, so no index is needed to make it fast. `dstrack log ` - is also accepted, resolved to a `dataset_id` via the same path-matching used by - `dstrack track` (see [Path resolution](#path-resolution-relative-root-overridable-never-a-shared-pointer)); if - the path doesn't exactly match any dataset's last recorded one, `--dataset-id` is - required instead. +Two different jobs are served by one mechanism: + +- **History of one dataset** (`dstrack log `) needs that dataset's + lineage, cheaply. `dstrack log ` is also accepted, resolved to a `dataset_id` + via the same path-matching used by `dstrack track` (see + [Path resolution](#path-resolution-relative-root-overridable-never-a-shared-pointer)); if + the path doesn't exactly match any dataset's last recorded one, the `dataset_id` must + be passed instead. - **Search across many datasets** (`dstrack search ...`, by name/tag/pipeline stage) would otherwise mean opening every dataset's snapshot JSON just to read its name or tags, files that may also contain large histograms and MinHash/HyperLogLog sketches @@ -95,15 +95,24 @@ never drift apart. Anyone who clones or pulls the repo has identical, correct history immediately, with no rebuild step. -`cache/index.db` (SQLite, stdlib `sqlite3`, no new dependency) is a pure performance -accelerator for cross-dataset search, built by reading the small `log.jsonl` files, -never the heavy snapshot JSON. It is **not** a second source of truth: it is gitignored -and safe to delete at any time. Every `dstrack track` call appends its new row -straight to the index, and every `dstrack search` also stats each dataset's `log.jsonl` -first and re-syncs any lines it's missing (size/mtime tracked per dataset), so -first-use, deletion, and pulls from other machines are all covered too. Between the -two, a search can never see stale results, only, in the worst case, a slower one while -it resyncs. +`.cache/index.db` (SQLite, stdlib `sqlite3`, no new dependency) is a pure performance +accelerator, built by reading the small `log.jsonl` files, never the heavy snapshot +JSON. Both `dstrack log` and `dstrack search` query it. It is **not** a second source of +truth: it is gitignored and safe to delete at any time. + +Reads re-sync before querying: each command stats every dataset's `log.jsonl` first +(size/mtime tracked per dataset) and reimports the ones that changed, so first-use, +deletion, and pulls from other machines are all covered. `dstrack track` therefore does +not need to write to the index at all. A query can never see stale results, only, in the +worst case, a slower one while it resyncs. + +**History is reachability, not file order.** `HEAD` is written after the `log.jsonl` +append, so a crash between the two leaves an entry the store never committed, and the +*next* successful `track` appends after it, stranding it mid-file. Reading the log +top-to-bottom, or up to the `HEAD` line, would both present that entry as real history. +So a dataset's history is derived by walking `parent_snapshot_id` back from `HEAD`, and +entries that walk does not reach are ignored. This holds for the index too, which is +built from the log and therefore inherits its unreachable entries. ### Path resolution: relative, root-overridable, never a shared pointer Every snapshot's `dataset_path` (ADR-0001) is stored relative to a *path root*, always @@ -192,7 +201,7 @@ is the supported answer. | `datasets//snapshots/*.json` | Yes | Source of truth for snapshot content | | `datasets//log.jsonl` | Yes | Source of truth for history; small and append-only | | `datasets//HEAD` | Yes | Small pointer; part of the same atomic write as the above | -| `cache/index.db` | No | Disposable, rebuildable search index | +| `.cache/index.db` | No | Disposable, rebuildable query index | ## Example walkthrough @@ -257,13 +266,14 @@ recorded in `log.jsonl`, not reconciled against anyone else's path. Their next snapshot of that same file can drop `--dataset-id`, since it will now match their own machine's most recent recorded path. -Finally, both `dstrack log 8f14e45f-...` (reads `log.jsonl` directly, three entries, -no `cache/index.db` involved) and `dstrack search --tag pipeline_stage=raw` (built from -`cache/index.db`, rebuilt from `log.jsonl` if the cache is missing or was never built on -this machine) return the same answer regardless of which machine ran them. +Finally, both `dstrack log 8f14e45f-...` (three entries, walked back from `HEAD`) and +`dstrack search --tag pipeline_stage=raw` return the same answer regardless of which +machine ran them: both query `.cache/index.db`, which is built from `log.jsonl` and +re-synced from it on every read, so a machine that has never built the index and a +machine whose index predates a `git pull` both still answer from the committed logs. ## Consequences -- `.dstrack/` is committed to git by default; only `cache/` is gitignored. Cloning the +- `.dstrack/` is committed to git by default; only `.cache/` is gitignored. Cloning the repo is sufficient to get full, correct dataset history, no rebuild, fetch, or config step required. - Dataset identity is a permanent `dataset_id` (UUID4), independent of both the file's @@ -279,8 +289,11 @@ this machine) return the same answer regardless of which machine ran them. - `log.jsonl` and `snapshots/*.json` must be written together, atomically, by the same operation. A store writer that fails to keep them in sync produces an inconsistent dataset history; this is an implementation invariant, not just a convention. -- `cache/index.db` can be deleted at any time with no data loss, only a slower next - search until it's rebuilt. +- `.cache/index.db` can be deleted at any time with no data loss, only a slower next + query until it's rebuilt. +- A dataset's history is what is reachable from `HEAD`, never what the log file happens + to contain. Any reader that trusts file order will surface snapshots the store never + committed; this is an implementation invariant, not a convention. - No reader information is ever written to the store, and no future field may reintroduce it: because `.dstrack/` is committed, an importable path read back out of it would execute code chosen by whoever authored the commit. Readers reach `dstrack` from the diff --git a/docs/gen_logo.py b/docs/gen_logo.py new file mode 100644 index 0000000..ffebbb5 --- /dev/null +++ b/docs/gen_logo.py @@ -0,0 +1,389 @@ +"""Regenerate docs/assets/dstrack-logo.svg and .png. + +Renders the rest pose of the (corrected) slab geometry from +docs/javascripts/logo-animation.js at 1x and emits: + - the SVG as per-row run-length rects (same format as the original), + - the PNG at 17x on a 510x510 transparent canvas, art centered. +""" + +import math +import struct +import zlib +from typing import TypedDict + +# Grid and projection constants, in logo pixel units. They must stay in sync +# with the ones in logo-animation.js, or the animation's first frame will not +# match the SVG this script emits. +GRID_W, GRID_H = 24, 29 +CENTER_X, RADIUS, ISO_Y, THICKNESS = 12, 11, 0.5, 2 +EPS = 1e-9 +COS45 = math.sqrt(0.5) + +RGB = tuple[int, int, int] +#: The rendered artwork: one row per logo pixel, `None` where nothing is drawn. +Grid = list[list[RGB | None]] + + +class Layer(TypedDict): + """One slab: where its top face sits vertically, and its palette.""" + + cy: float + top: RGB + left: RGB + right: RGB + edge: RGB + + +class Edge(TypedDict): + """One segment of a slab's plan-view outline. + + `a`/`b` index into the outline points, `normal` is the outward normal at + rest, `seam` marks the dark corner tips, and `base` is the rest normal of + the smooth square face the segment belongs to. + """ + + a: int + b: int + normal: float + seam: bool + base: float + + +def hexc(s: str) -> RGB: + """Parse a "#rrggbb" color string. + + Args: + s: A hex color, leading "#" included. + + Returns: + The red, green and blue channels, each 0-255. + """ + return (int(s[1:3], 16), int(s[3:5], 16), int(s[5:7], 16)) + + +# One entry per slab, top to bottom: vertical center of the top face (screen +# units) and the palette taken verbatim from the SVG logo. `left`/`right` shade +# the side faces; `edge` is the dark seam painted on the corner tips. +LAYERS: list[Layer] = [ + Layer( + cy=6.5, + top=hexc("#a99cff"), + left=hexc("#7060e0"), + right=hexc("#4a3cc0"), + edge=hexc("#2f2490"), + ), + Layer( + cy=13.5, + top=hexc("#8b7bf0"), + left=hexc("#5b4bd6"), + right=hexc("#3a2fa6"), + edge=hexc("#241b6b"), + ), + Layer( + cy=20.5, + top=hexc("#6a5be0"), + left=hexc("#463aae"), + right=hexc("#2c2185"), + edge=hexc("#1d1568"), + ), +] + + +def mix(a: RGB, b: RGB, t: float) -> RGB: + """Linearly interpolate between two colors. + + Args: + a: The color returned at `t` = 0. + b: The color returned at `t` = 1. + t: The blend position, expected in [0, 1] (not clamped here). + + Returns: + The blended color, rounded per channel. + """ + return tuple(round(a[i] + (b[i] - a[i]) * t) for i in range(3)) + + +def build_outline() -> tuple[list[float], list[float], list[Edge], int]: + """Build the top-face outline of a slab in plan view. + + The outline is a 45deg-rotated square whose edges are staircases of 2x2 + plan-unit steps, walked as a closed loop: right side bottom-to-top, then + left side top-to-bottom. Every vertex lands on a logo pixel boundary, which + is what lets the rendered result reproduce the SVG exactly. + + Returns: + The outline point coordinates `px` and `py`, the `edges` connecting + them in order, and the point count `n`. + """ + pts: list[tuple[float, float]] = [] + for dy in range(-5, 6): + hw = RADIUS - 2 * abs(dy) + pts += [(hw, dy - 0.5), (hw, dy + 0.5)] + for dy in range(5, -6, -1): + hw = RADIUS - 2 * abs(dy) + pts += [(-hw, dy + 0.5), (-hw, dy - 0.5)] + n = len(pts) + px = [p[0] for p in pts] + py = [p[1] / ISO_Y for p in pts] # undo the isometric foreshortening + edges: list[Edge] = [] + for i in range(n): + j = (i + 1) % n + mx, my = (px[i] + px[j]) / 2, (py[i] + py[j]) / 2 + nx, ny = py[j] - py[i], -(px[j] - px[i]) + # The outline is convex about the origin, so the normal pointing away + # from the center is the outward one. + if nx * mx + ny * my < 0: + nx, ny = -nx, -ny + # Corner tips sit exactly on a plan axis; everything else belongs to + # the face whose rest normal is the nearest odd multiple of 45deg. + mid = math.atan2(my, mx) + quadrant = mid / (math.pi / 2) + seam = abs(quadrant - round(quadrant)) < 1e-6 + m = mid % (2 * math.pi) + base = math.floor(m / (math.pi / 2)) * (math.pi / 2) + math.pi / 4 + edges.append(Edge(a=i, b=j, normal=math.atan2(ny, nx), seam=seam, base=base)) + return px, py, edges, n + + +OUT_PX, OUT_PY, OUT_EDGES, OUT_N = build_outline() + + +def shade(layer: Layer, normal_angle: float) -> RGB: + """Pick the side-face color for a face pointing in a given direction. + + Fixed light from the left: darkest when a face points right (rest angle + 45deg), lightest when it points left (135deg), clamped beyond, so the + resting pose uses the exact logo colors. + + Args: + layer: The slab being drawn, supplying the palette to blend. + normal_angle: The face's current outward normal, in radians. + + Returns: + The blend of the layer's right and left face colors for that angle. + """ + t = min(1, max(0, (COS45 - math.cos(normal_angle)) / (2 * COS45))) + return mix(layer["right"], layer["left"], t) + + +def fill_polygon(grid: Grid, xs: list[float], ys: list[float], rgb: RGB) -> None: + """Paint a solid color over a (possibly non-convex) polygon. + + Even-odd scanline fill in grid coordinates: a cell is filled when its + center lies inside the polygon. Later fills overwrite earlier ones. + + Args: + grid: The target grid, modified in place. + xs: The polygon vertices' horizontal coordinates. + ys: The polygon vertices' vertical coordinates, paired with `xs`. + rgb: The color to paint. + """ + count = len(xs) + r0 = max(0, math.ceil(min(ys) - 0.5)) + r1 = min(GRID_H - 1, math.floor(max(ys) - 0.5)) + for r in range(r0, r1 + 1): + yc = r + 0.5 + # Collect where the scanline crosses each edge; sorted pairs of + # crossings bracket the spans that are inside the polygon. + crossings: list[float] = [] + for i in range(count): + j = (i + 1) % count + y1, y2 = ys[i], ys[j] + if (y1 <= yc) == (y2 <= yc): + continue + crossings.append(xs[i] + ((yc - y1) / (y2 - y1)) * (xs[j] - xs[i])) + crossings.sort() + for k in range(0, len(crossings) - 1, 2): + c0 = max(0, math.ceil(crossings[k] - 0.5)) + c1 = min(GRID_W - 1, math.floor(crossings[k + 1] - 0.5)) + for c in range(c0, c1 + 1): + grid[r][c] = rgb + + +def draw_layer(grid: Grid, layer: Layer, angle: float) -> None: + """Rasterize one slab, side faces then top face. + + Args: + grid: The target grid, modified in place. + layer: The slab to draw, supplying its height and palette. + angle: The slab's rotation in the plan plane, in radians. + """ + cos_a, sin_a = math.cos(angle), math.sin(angle) + # Rotate the outline in plan view, then project to screen: x straight + # across, y foreshortened onto the top face, extruded down by THICKNESS. + sx: list[float] = [] + syt: list[float] = [] + syb: list[float] = [] + for i in range(OUT_N): + x = OUT_PX[i] * cos_a - OUT_PY[i] * sin_a + y = OUT_PX[i] * sin_a + OUT_PY[i] * cos_a + sx.append(CENTER_X + x) + syt.append(layer["cy"] + y * ISO_Y) + syb.append(syt[-1] + THICKNESS) + # Side faces: every outline edge whose outward normal currently points + # toward the viewer. Shaded faces first, dark seams second so they stay + # visible at the corners, top face last so it owns the shared boundary + # pixels and overpaints any far-side step that slipped through. + for seam_pass in (False, True): + for e in OUT_EDGES: + if e["seam"] != seam_pass: + continue + if math.sin(e["normal"] + angle) <= EPS: + continue + a, b = e["a"], e["b"] + rgb = layer["edge"] if e["seam"] else shade(layer, e["base"] + angle) + fill_polygon( + grid, + [sx[a], sx[b], sx[b], sx[a]], + [syt[a], syt[b], syb[b], syb[a]], + rgb, + ) + fill_polygon(grid, sx, syt, layer["top"]) + + +def render_rest() -> Grid: + """Render the unrotated pose of the whole stack. + + Bottom slab first, so the ones above paint over it where they overlap. + + Returns: + A GRID_H x GRID_W grid holding the drawn colors. + """ + grid: Grid = [[None] * GRID_W for _ in range(GRID_H)] + for i in range(len(LAYERS) - 1, -1, -1): + draw_layer(grid, LAYERS[i], 0.0) + return grid + + +def to_hex(rgb: RGB) -> str: + """Format a color for SVG. + + Args: + rgb: The channel values to format. + + Returns: + The color as a "#rrggbb" string. + """ + return "#{:02x}{:02x}{:02x}".format(*rgb) + + +def emit_svg(grid: Grid, path: str, unit: int = 16) -> int: + """Write the grid out as an SVG. + + One rect per run of same-colored cells in a row (the format the original + hand-written logo used), on a viewBox of one unit per logo pixel. + + Args: + grid: The rendered artwork to serialize. + path: Where to write the SVG. + unit: Rendered pixels per logo pixel, setting the SVG's width/height. + + Returns: + The number of rects emitted. + """ + rects: list[str] = [] + for r in range(GRID_H): + c = 0 + while c < GRID_W: + if grid[r][c] is None: + c += 1 + continue + # Extend the run as far as the color holds, then emit it as one rect. + start, color = c, grid[r][c] + while c < GRID_W and grid[r][c] == color: + c += 1 + rects.append( + f'' + ) + svg = ( + f'\n' + "dstrack - logo\n" + "".join(rects) + "\n\n" + ) + with open(path, "w") as f: + f.write(svg) + return len(rects) + + +def emit_png(grid: Grid, path: str, size: int = 510, scale: int = 17) -> None: + """Write the grid out as a square, nearest-neighbor upscaled PNG. + + Hand-rolled RGBA PNG writer (no image dependency): the art is centered on a + transparent canvas and every cell becomes one solid block, so the pixel art + stays crisp. + + Args: + grid: The rendered artwork to serialize. + path: Where to write the PNG. + size: The canvas edge length, in output pixels. + scale: Output pixels per logo pixel; the art must fit within `size`. + """ + # Center the drawn content on the square canvas. + content_rows = [r for r in range(GRID_H) if any(grid[r])] + content_cols = [c for c in range(GRID_W) if any(row[c] for row in grid)] + art_w = (max(content_cols) - min(content_cols) + 1) * scale + art_h = (max(content_rows) - min(content_rows) + 1) * scale + ox = (size - art_w) // 2 - min(content_cols) * scale + oy = (size - art_h) // 2 - min(content_rows) * scale + + buf = bytearray(size * size * 4) + for r in range(GRID_H): + for c in range(GRID_W): + rgb = grid[r][c] + if rgb is None: + continue + for yy in range(oy + r * scale, oy + (r + 1) * scale): + row_base = (yy * size + ox + c * scale) * 4 + for i in range(scale): + buf[row_base + i * 4 : row_base + i * 4 + 4] = bytes((*rgb, 255)) + + # PNG scanlines are each prefixed with a filter-type byte; 0 means "none". + raw = b"".join( + b"\x00" + bytes(buf[y * size * 4 : (y + 1) * size * 4]) for y in range(size) + ) + + def chunk(typ: bytes, data: bytes) -> bytes: + """Frame a payload as a PNG chunk. + + Args: + typ: The four-byte chunk type, such as b"IHDR". + data: The chunk payload. + + Returns: + The chunk: length, type, payload, then CRC32 of type and payload. + """ + c = struct.pack(">I", len(data)) + typ + data + return c + struct.pack(">I", zlib.crc32(typ + data) & 0xFFFFFFFF) + + png = ( + b"\x89PNG\r\n\x1a\n" + # 8 bits per channel, color type 6 (RGBA), default compression/filter/interlace. + + chunk(b"IHDR", struct.pack(">IIBBBBB", size, size, 8, 6, 0, 0, 0)) + + chunk(b"IDAT", zlib.compress(raw, 9)) + + chunk(b"IEND", b"") + ) + with open(path, "wb") as f: + f.write(png) + + +def main() -> None: + """Render the rest pose and write the SVG, logo PNG and favicon PNG.""" + grid = render_rest() + # Symmetry check on the artifact itself: every row's drawn extent must be + # centered on x = CENTER_X. + for r in range(GRID_H): + cols = [c for c in range(GRID_W) if grid[r][c] is not None] + if cols: + assert min(cols) + max(cols) == 2 * CENTER_X - 1, ( + f"row {r} asymmetric: {min(cols)}..{max(cols)}" + ) + n = emit_svg(grid, "docs/assets/dstrack-logo.svg") + emit_png(grid, "docs/assets/dstrack-logo.png", size=510, scale=17) + emit_png(grid, "docs/assets/dstrack-favicon.png", size=60, scale=2) + widths = [sum(1 for c in range(GRID_W) if grid[r][c]) for r in range(GRID_H)] + print(f"SVG written ({n} rects), logo PNG (510x510 @17x), favicon PNG (60x60 @2x)") + print("row widths:", widths) + + +if __name__ == "__main__": + main() diff --git a/docs/getting_started.md b/docs/getting_started.md index a005b93..5f39ffc 100644 --- a/docs/getting_started.md +++ b/docs/getting_started.md @@ -120,21 +120,66 @@ For more information on what `dstrack track` allows you to do, simply ask for th dstrack track --help ``` +## 4. View a dataset's history + +`dstrack log` shows a dataset's snapshots as a timeline, newest first. Point it at the +dataset's path, and `dstrack` works out which dataset that is: + +```bash +dstrack log data.csv +``` + +```text +● a587afd9 Customers HEAD +│ 2 minutes ago by you +│ 4 rows +1 3 cols +│ data.csv +│ +● 605eb940 Customers +│ 5 minutes ago by you +│ 3 rows +1 3 cols +│ data.csv +│ +● 0c82297f Customers + 1 hour ago by you + 2 rows 3 cols + data.csv +``` + +Each node is one snapshot, showing how its row and column counts changed from the +snapshot below it. `HEAD` marks the latest. + +This is also how you find the `--dataset-id` the rename above needs: pass the path while +it still matches, or list what the store knows by asking for a dataset that isn't there. +A dataset id works anywhere a path does, and keeps working after the file is renamed, +moved, or deleted: + +```bash +dstrack log +``` + +Useful flags: `-n 5` to show only the latest few, `--oneline` to condense each snapshot +to a single line, and `--reverse` to read oldest-first. + ## Where snapshots live Each snapshot is a JSON file under its dataset's directory: ```text .dstrack/ +├── .cache/ +│ └── index.db # not committed; rebuilt on demand └── datasets/ └── / + ├── HEAD # the latest snapshot + ├── log.jsonl # append-only history, one line per snapshot └── snapshots/ └── .json ``` -The store also keeps an append-only log and a `HEAD` pointer per dataset. `.dstrack/` is -plain text and safe to commit alongside your code, giving you a versioned audit trail of -how your datasets evolved. +`.dstrack/` is plain text and safe to commit alongside your code, giving you a versioned +audit trail of how your datasets evolved. The exception is `.cache/`, which holds only +what can be derived from the logs beside it, and is gitignored for you. ## Benchmarking (optional) diff --git a/docs/index.md b/docs/index.md index a850214..5facd44 100644 --- a/docs/index.md +++ b/docs/index.md @@ -5,7 +5,7 @@ icon: lucide/rocket # dstrack
- ![dstrack-logo](assets/dstrack-logo.png){ width="200" } + ![dstrack-logo](assets/dstrack-logo.png){ width="200" #dstrack-logo }
@@ -13,6 +13,20 @@ icon: lucide/rocket `dstrack` helps data scientists and ML engineers track how datasets evolve over time - catching schema drift, distribution shifts, and unexpected mutations before they silently break pipelines or degrade model performance. +## Why dstrack? + +Data pipelines break silently. A column gets renamed upstream, a vendor changes a file format, or a feature distribution shifts after a data refresh - and you only find out when model accuracy drops in production. + +`dstrack` gives you an audit trail for your datasets so you can catch these problems early, understand what changed, and help you develop your datasets faster. + +## What dstrack is (and isn't) + +`dstrack` records *semantics* about a dataset - its schema, a content fingerprint, and per-column statistics - not the dataset itself. +It is not a data version control system: unlike DVC, LakeFS, or Git LFS, it never copies your files into a store or pushes them to a +remote, and it cannot check an old version of `data.csv` back out for you. Your data stays wherever it already lives, and `dstrack` keeps a small, +human-readable audit trail beside your code so you can see how that data changed and when. If you need to store and retrieve the bytes, reach for +DVC and use `dstrack` alongside it. + ## Features - **Dataset versioning** - snapshot a dataset and track its lineage across pipeline stages @@ -50,12 +64,6 @@ dstrack track data.csv New here? The [Getting Started guide](getting_started.md) walks through the whole flow step by step. -## Why dstrack? - -Data pipelines break silently. A column gets renamed upstream, a vendor changes a file format, or a feature distribution shifts after a data refresh - and you only find out when model accuracy drops in production. - -`dstrack` gives you an audit trail for your datasets so you can catch these problems early, understand what changed, and reproduce any past state of your data. - ## License `dstrack` is distributed under the [MIT License](https://github.com/leoyala/dstrack/blob/main/LICENSE). diff --git a/docs/javascripts/logo-animation.js b/docs/javascripts/logo-animation.js new file mode 100644 index 0000000..7071246 --- /dev/null +++ b/docs/javascripts/logo-animation.js @@ -0,0 +1,310 @@ +/* dstrack logo animation. + * + * Re-renders the pixel-art isometric logo (three stacked square slabs seen + * from an isometric viewpoint) on a small canvas that is upscaled with + * `image-rendering: pixelated`. At scroll position 0 the render is + * pixel-identical to docs/assets/dstrack-logo.svg; as the user scrolls down, + * each slab rotates in the horizontal plane at its own speed and direction. + * + * The chunky stair steps on each slab's border are modeled as real geometry + * (a stepped polygon in plan view), not left to rasterization: this keeps the + * number of border indentations constant while a slab turns. The scene is + * rasterized at SCALE x the logo grid so the rotating steps stay steady + * instead of crawling. + * + * The script progressively enhances the landing page: it looks for the + * `#dstrack-logo` image and swaps it for a canvas. Without JavaScript the + * original PNG logo is shown. Re-initializes after instant navigation + * (zensical swaps the page content via XHR) and honors + * `prefers-reduced-motion`. + */ +(() => { + "use strict"; + + // Grid and projection constants, in logo pixel units. The slab is a + // 4-fold-symmetric stepped square in plan view: every corner tip step is + // 2 plan units wide, front/back and left/right alike. + const GRID_W = 24; + const GRID_H = 29; + const CENTER_X = 12; // horizontal center of the slabs + const RADIUS = 11; // half-diagonal of each square slab (plan view) + const ISO_Y = 0.5; // 2:1 isometric vertical foreshortening + const THICKNESS = 2; // slab thickness + const SCALE = 4; // internal subpixels per logo pixel + const CANVAS_W = GRID_W * SCALE; + const CANVAS_H = GRID_H * SCALE; + const EPS = 1e-9; + const COS45 = Math.SQRT1_2; + + // Rotation per scrolled pixel (radians), top slab first. Alternating + // directions make the stack feel like meshing gears. + const SPEEDS = [0.36, -0.24, 0.16].map((deg) => (deg * Math.PI) / 180); + const EASE = 0.12; // per-frame easing toward the scroll target + + const hex = (s) => [ + parseInt(s.slice(1, 3), 16), + parseInt(s.slice(3, 5), 16), + parseInt(s.slice(5, 7), 16), + ]; + + // One entry per slab, top to bottom: vertical center of the top face + // (screen units) and the palette taken verbatim from the SVG logo. + // `left`/`right` shade the side faces at rest; `edge` is the dark seam + // painted on the corner tips. + const LAYERS = [ + { cy: 6.5, top: hex("#a99cff"), left: hex("#7060e0"), right: hex("#4a3cc0"), edge: hex("#2f2490") }, + { cy: 13.5, top: hex("#8b7bf0"), left: hex("#5b4bd6"), right: hex("#3a2fa6"), edge: hex("#241b6b") }, + { cy: 20.5, top: hex("#6a5be0"), left: hex("#463aae"), right: hex("#2c2185"), edge: hex("#1d1568") }, + ]; + + const mix = (a, b, t) => [ + Math.round(a[0] + (b[0] - a[0]) * t), + Math.round(a[1] + (b[1] - a[1]) * t), + Math.round(a[2] + (b[2] - a[2]) * t), + ]; + + // ------------------------------------------------------------------ + // Slab geometry + // ------------------------------------------------------------------ + + // Top-face outline in plan view (horizontal plane, +y toward the viewer), + // built once from the pixel pattern of the logo: a 45deg-rotated square + // whose edges are staircases of 2x2 plan-unit steps. At rest every vertex + // projects onto a logo pixel boundary, which is what makes the resting + // frame reproduce the SVG exactly. + // + // Each outline edge knows its outward normal (rest angle), whether it is a + // dark corner-tip seam, and which face of the underlying smooth square it + // belongs to (rest normal of that face), used for shading. + function buildOutline() { + const pts = []; + for (let dy = -5; dy <= 5; dy++) { + const hw = RADIUS - 2 * Math.abs(dy); + pts.push([hw, dy - 0.5], [hw, dy + 0.5]); + } + for (let dy = 5; dy >= -5; dy--) { + const hw = RADIUS - 2 * Math.abs(dy); + pts.push([-hw, dy + 0.5], [-hw, dy - 0.5]); + } + const n = pts.length; + const px = new Float64Array(n); + const py = new Float64Array(n); + for (let i = 0; i < n; i++) { + px[i] = pts[i][0]; + py[i] = pts[i][1] / ISO_Y; // undo the isometric foreshortening + } + const edges = []; + for (let i = 0; i < n; i++) { + const j = (i + 1) % n; + const mx = (px[i] + px[j]) / 2; + const my = (py[i] + py[j]) / 2; + let nx = py[j] - py[i]; + let ny = -(px[j] - px[i]); + if (nx * mx + ny * my < 0) { + nx = -nx; + ny = -ny; + } + // Corner tips sit exactly on a plan axis; everything else belongs to + // the face whose rest normal is the nearest odd multiple of 45deg. + const mid = Math.atan2(my, mx); + const quadrant = mid / (Math.PI / 2); + const seam = Math.abs(quadrant - Math.round(quadrant)) < 1e-6; + const m = ((mid % (2 * Math.PI)) + 2 * Math.PI) % (2 * Math.PI); + const base = Math.floor(m / (Math.PI / 2)) * (Math.PI / 2) + Math.PI / 4; + edges.push({ a: i, b: j, normal: Math.atan2(ny, nx), seam, base }); + } + return { px, py, edges, count: n }; + } + + const OUTLINE = buildOutline(); + + // Fixed light from the left: the darkest shade when a face points right + // (rest angle 45deg), the lightest when it points left (135deg), clamped + // beyond, so the resting pose uses the exact logo colors. + function shade(layer, normalAngle) { + const t = Math.min(1, Math.max(0, (COS45 - Math.cos(normalAngle)) / (2 * COS45))); + return mix(layer.right, layer.left, t); + } + + // ------------------------------------------------------------------ + // Rasterization + // ------------------------------------------------------------------ + + function setPixel(data, x, y, rgb) { + const i = (y * CANVAS_W + x) * 4; + data[i] = rgb[0]; + data[i + 1] = rgb[1]; + data[i + 2] = rgb[2]; + data[i + 3] = 255; + } + + // Even-odd scanline fill of a (possibly non-convex) polygon given in + // canvas subpixel coordinates. A pixel is filled when its center lies + // inside the polygon. + function fillPolygon(data, xs, ys, count, rgb) { + let minY = Infinity; + let maxY = -Infinity; + for (let i = 0; i < count; i++) { + minY = Math.min(minY, ys[i]); + maxY = Math.max(maxY, ys[i]); + } + const r0 = Math.max(0, Math.ceil(minY - 0.5)); + const r1 = Math.min(CANVAS_H - 1, Math.floor(maxY - 0.5)); + const crossings = []; + for (let r = r0; r <= r1; r++) { + const yc = r + 0.5; + crossings.length = 0; + for (let i = 0; i < count; i++) { + const j = (i + 1) % count; + const y1 = ys[i]; + const y2 = ys[j]; + if ((y1 <= yc) === (y2 <= yc)) continue; + crossings.push(xs[i] + ((yc - y1) / (y2 - y1)) * (xs[j] - xs[i])); + } + crossings.sort((a, b) => a - b); + for (let k = 0; k + 1 < crossings.length; k += 2) { + const c0 = Math.max(0, Math.ceil(crossings[k] - 0.5)); + const c1 = Math.min(CANVAS_W - 1, Math.floor(crossings[k + 1] - 0.5)); + for (let c = c0; c <= c1; c++) setPixel(data, c, r, rgb); + } + } + } + + const quadX = new Float64Array(4); + const quadY = new Float64Array(4); + + function drawLayer(data, layer, angle) { + const cosA = Math.cos(angle); + const sinA = Math.sin(angle); + const n = OUTLINE.count; + const sx = new Float64Array(n); + const syT = new Float64Array(n); + const syB = new Float64Array(n); + for (let i = 0; i < n; i++) { + const x = OUTLINE.px[i] * cosA - OUTLINE.py[i] * sinA; + const y = OUTLINE.px[i] * sinA + OUTLINE.py[i] * cosA; + sx[i] = (CENTER_X + x) * SCALE; + syT[i] = (layer.cy + y * ISO_Y) * SCALE; + syB[i] = syT[i] + THICKNESS * SCALE; + } + + // Side faces: every outline edge whose outward normal currently points + // toward the viewer, extruded down by the slab thickness. Shaded faces + // first, dark seams second so they stay visible at the corners, top + // face last so it owns the shared boundary pixels. Far-side steps that + // slip through the visibility test are overpainted by the top face. + for (const pass of [false, true]) { + for (const e of OUTLINE.edges) { + if (e.seam !== pass) continue; + if (Math.sin(e.normal + angle) <= EPS) continue; + quadX[0] = sx[e.a]; + quadX[1] = sx[e.b]; + quadX[2] = sx[e.b]; + quadX[3] = sx[e.a]; + quadY[0] = syT[e.a]; + quadY[1] = syT[e.b]; + quadY[2] = syB[e.b]; + quadY[3] = syB[e.a]; + const rgb = e.seam ? layer.edge : shade(layer, e.base + angle); + fillPolygon(data, quadX, quadY, 4, rgb); + } + } + fillPolygon(data, sx, syT, n, layer.top); + } + + function render(imageData, angles) { + imageData.data.fill(0); + for (let i = LAYERS.length - 1; i >= 0; i--) { + drawLayer(imageData.data, LAYERS[i], angles[i]); + } + } + + // ------------------------------------------------------------------ + // Page wiring + // ------------------------------------------------------------------ + + const reduceMotion = window.matchMedia("(prefers-reduced-motion: reduce)"); + + let canvas = null; + let ctx = null; + let imageData = null; + let rafId = 0; + const current = [0, 0, 0]; + const target = [0, 0, 0]; + + function renderNow() { + render(imageData, current); + ctx.putImageData(imageData, 0, 0); + } + + function updateTarget() { + const y = reduceMotion.matches ? 0 : window.scrollY || 0; + for (let i = 0; i < 3; i++) target[i] = y * SPEEDS[i]; + } + + function tick() { + let settled = true; + let changed = false; + for (let i = 0; i < 3; i++) { + const delta = target[i] - current[i]; + if (Math.abs(delta) < 1e-4) { + if (current[i] !== target[i]) { + current[i] = target[i]; + changed = true; + } + } else { + current[i] += delta * EASE; + changed = true; + settled = false; + } + } + if (changed && canvas && canvas.isConnected) renderNow(); + rafId = settled ? 0 : requestAnimationFrame(tick); + } + + function onScroll() { + if (!canvas || !canvas.isConnected) return; + updateTarget(); + if (!rafId) rafId = requestAnimationFrame(tick); + } + + function initLogo() { + const img = document.getElementById("dstrack-logo"); + if (!img) return; + const replacement = document.createElement("canvas"); + replacement.width = CANVAS_W; + replacement.height = CANVAS_H; + replacement.className = img.className; + replacement.setAttribute("role", "img"); + replacement.setAttribute("aria-label", img.alt || "dstrack logo"); + replacement.style.width = (img.getAttribute("width") || "200") + "px"; + replacement.style.height = "auto"; + replacement.style.aspectRatio = `${GRID_W} / ${GRID_H}`; + replacement.style.imageRendering = "pixelated"; + img.replaceWith(replacement); + + canvas = replacement; + ctx = canvas.getContext("2d"); + imageData = ctx.createImageData(CANVAS_W, CANVAS_H); + updateTarget(); + for (let i = 0; i < 3; i++) current[i] = target[i]; + renderNow(); + } + + function boot() { + initLogo(); + // Instant navigation swaps the page content via XHR without a reload; + // watch for the logo image to (re)appear and upgrade it again. + new MutationObserver(() => { + if (!canvas || !canvas.isConnected) initLogo(); + }).observe(document.documentElement, { childList: true, subtree: true }); + window.addEventListener("scroll", onScroll, { passive: true }); + reduceMotion.addEventListener("change", onScroll); + } + + if (document.readyState === "loading") { + document.addEventListener("DOMContentLoaded", boot); + } else { + boot(); + } +})(); diff --git a/docs/roadmap.md b/docs/roadmap.md index 4b9cfd5..129b768 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -25,7 +25,7 @@ Foundation for all future capabilities: reading a dataset and capturing an immut - [x] Schema extraction (column names and types) - [x] Per-column statistics (counts, ranges, null rates, etc.) - [x] `dstrack init` and `dstrack track` commands -- [ ] `dstrack log` command +- [x] `dstrack log` command ## v0.2 - Diff Engine diff --git a/src/dstrack/_cli.py b/src/dstrack/_cli.py index 070db64..02af29b 100644 --- a/src/dstrack/_cli.py +++ b/src/dstrack/_cli.py @@ -6,6 +6,7 @@ import typer from dstrack import __version__, console +from dstrack._log import log from dstrack._store_template import STORE_TEMPLATE, materialize from dstrack._track import track from dstrack.errors import StoreInitError @@ -89,6 +90,11 @@ def init( track ) +app.command( + help="Show the snapshot history of a tracked dataset, given its dataset id " + "or its path." +)(log) + @app.command(help="Print package version.") def version() -> None: diff --git a/src/dstrack/_log.py b/src/dstrack/_log.py new file mode 100644 index 0000000..68efe15 --- /dev/null +++ b/src/dstrack/_log.py @@ -0,0 +1,268 @@ +"""The ``dstrack log`` command: show a tracked dataset's history.""" + +import os +import sqlite3 +import sys +import uuid +from datetime import UTC, datetime +from itertools import pairwise +from pathlib import Path +from typing import Annotated + +import typer + +from dstrack import cache, console, store +from dstrack._log_render import render_history +from dstrack.errors import ( + DatasetNotFoundError, + StoreCorruptionError, + StoreNotFoundError, +) +from dstrack.paths import resolve_store_root + + +def log( + target: Annotated[ + str, + typer.Argument( + metavar="TARGET", + help="Dataset to show the history of: either a dataset id, or the " + "path to a tracked dataset file.", + ), + ], + limit: Annotated[ + int | None, + typer.Option( + "-n", + "--limit", + min=1, + help="Show at most this many snapshots, counting back from the latest.", + ), + ] = None, + reverse: Annotated[ + bool, + typer.Option("--reverse", help="Show oldest first instead of newest first."), + ] = False, + oneline: Annotated[ + bool, + typer.Option("--oneline", help="Condense each snapshot to a single line."), + ] = False, + yes: Annotated[ + bool, + typer.Option( + "-y", + "--yes", + help="Build the snapshot index without asking, if it does not exist yet.", + ), + ] = False, + rebuild: Annotated[ + bool, + typer.Option( + "--rebuild", + help="Discard the snapshot index and rebuild it from the dataset logs.", + ), + ] = False, + root: Annotated[ + Path | None, + typer.Option( + "--root", + help="Path root a TARGET path is made relative to, matching the one " + "`dstrack track` recorded it with. Defaults to the store root (the " + "directory containing .dstrack/). Ignored when TARGET is a dataset id.", + ), + ] = None, +) -> None: + """Show the snapshot history of a tracked dataset.""" + # Locate the .dstrack/ store; without one there is no history to show. + try: + store_root = resolve_store_root() + except StoreNotFoundError as e: + console.error(str(e)) + raise typer.Exit(code=1) from e + + _ensure_index(store_root, yes=yes, rebuild=rebuild) + + try: + # Sync before resolving: the index does not travel with the repository, + # so a dataset tracked elsewhere is only visible once its log is read. + reason = cache.sync(store_root) + if reason is not None: + console.warning(f"Rebuilt the snapshot index: {reason}.") + + dataset_id = _resolve_target(target, store_root=store_root, root=root) + history = cache.query_history(dataset_id, store_root=store_root) + except ( + DatasetNotFoundError, + StoreCorruptionError, + ValueError, + OSError, + sqlite3.DatabaseError, + ) as e: + console.error(str(e)) + raise typer.Exit(code=1) from e + + if not history: + console.warning(f"Dataset {dataset_id} has no snapshots yet.") + return + + # Deltas are keyed by lineage, not by display position, so that slicing and + # reversing below cannot change what a snapshot is compared against. + parents = {entry.snapshot_id: parent for entry, parent in pairwise(history)} + head = history[0].snapshot_id + + # Limit counts back from HEAD, then --reverse flips what is shown; the two + # compose the way `git log -n ... --reverse` does. + shown = history if limit is None else history[:limit] + if reverse: + shown = list(reversed(shown)) + + console.display( + render_history( + shown, + parents=parents, + head_id=head, + now=datetime.now(UTC), + oneline=oneline, + ) + ) + + +def _ensure_index(store_root: Path, *, yes: bool, rebuild: bool) -> None: + """Make sure the snapshot index may be built, asking first if need be. + + Building parses every dataset's log, so it is offered rather than done + silently. Consent is only about the first build: an index that exists but + cannot be used is rebuilt without asking, since it holds nothing that the + logs do not. + + Args: + store_root: Path to the ``.dstrack/`` directory. + yes: Consent given up front, so nothing is asked. + rebuild: Discard any existing index, which implies consent. + + Raises: + typer.Exit: Code 0 if the user declines to build it; code 1 if there + is no index and no terminal to ask at. + """ + path = cache.index_path(store_root) + if rebuild: + path.unlink(missing_ok=True) + console.info("Rebuilding the snapshot index from the dataset logs...") + return + if cache.index_exists(store_root): + return + + if not yes: + if not sys.stdin.isatty(): + console.error( + f"No snapshot index at `{path}`, and there is no terminal to " + "ask whether to build one. Re-run with `--yes` to build it." + ) + raise typer.Exit(code=1) + console.info(f"No snapshot index at `{path}` yet.") + if not typer.confirm("Build it now from the dataset logs?"): + console.warning( + "Not building the index; there is nothing to show without it." + ) + raise typer.Exit(code=0) + + console.info("Building the snapshot index from the dataset logs...") + + +def _resolve_target(target: str, *, store_root: Path, root: Path | None) -> str: + """Resolve a user-supplied dataset id or path to a dataset id. + + A ``dataset_id`` is a random UUID4, so nothing about it can be + derived from a path: a path is resolved by matching it against what each + dataset last recorded. `target` is read as an id first and as a path + second, but only when the id names a dataset that exists. Reading it as a + path first would let any file in the current directory shadow a real id, + and would send a mistyped path into the id lookup, which can only report + that no dataset goes by that name. + + Args: + target: A dataset id, or a path to a tracked dataset file. + store_root: Path to the ``.dstrack/`` directory. + root: Path root a `target` path is made relative to, or ``None`` to + use the store root. + + Returns: + The id of the dataset `target` names. + + Raises: + DatasetNotFoundError: If `target` is an id no dataset goes by, or a + path no dataset last recorded. + ValueError: If `target` and the path root are on different drives, so + no relative path exists between them. + """ + canonical = _canonical_uuid(target) + if canonical is not None and store.dataset_exists(canonical, store_root=store_root): + return canonical + + path_root = root if root is not None else store_root.parent + dataset_path = Path( + os.path.relpath(Path(target).resolve(), Path(path_root).resolve()) + ).as_posix() + + matched = cache.find_dataset_by_path(dataset_path, store_root=store_root) + if matched is not None: + return matched + + if canonical is not None: + raise DatasetNotFoundError( + f"No dataset {target!r} in the store.\n{_known_datasets(store_root)}" + ) + raise DatasetNotFoundError( + f"No tracked dataset at {target!r} (recorded as {dataset_path!r} " + f"relative to {path_root}).\n" + "If the file was renamed or moved, its recorded path no longer " + "matches; pass the dataset id instead. Use `--root` to change the " + "path root the recorded path is computed against.\n" + f"{_known_datasets(store_root)}" + ) + + +def _canonical_uuid(value: str) -> str | None: + """Return `value` as a canonical UUID string, or ``None`` if it is not one. + + Datasets are named by ``str(uuid.uuid4())``, but a user may paste an id + uppercased, braced, or without its dashes, all of which name the same + dataset. Canonicalizing accepts those spellings, and makes the result + provably free of path separators before it is joined onto the store. + + Args: + value: The string to interpret. + + Returns: + The canonical form of `value` if it is a UUID in any spelling, else + ``None``. + """ + try: + return str(uuid.UUID(value)) + except (ValueError, AttributeError, TypeError): + return None + + +def _known_datasets(store_root: Path) -> str: + """List the store's datasets, for an error message to end on. + + A dataset id says nothing on its own, so each is shown with the name and + path its latest snapshot recorded: that is what lets a reader recognize + the dataset they meant and copy its id. + + Args: + store_root: Path to the ``.dstrack/`` directory. + + Returns: + An indented listing of every dataset, or a note that there are none. + """ + datasets = cache.list_datasets(store_root=store_root) + lines = [ + f" {d.dataset_id} {(d.head.dataset_name if d.head else None) or '-'} " + f"{(d.head.dataset_path if d.head else None) or '-'}" + for d in datasets + ] + if not lines: + return "Known datasets:\n (none)\nRun `dstrack track ` to record one." + listing = "\n".join(lines) + return f"Known datasets:\n{listing}" diff --git a/src/dstrack/_log_render.py b/src/dstrack/_log_render.py new file mode 100644 index 0000000..354aa71 --- /dev/null +++ b/src/dstrack/_log_render.py @@ -0,0 +1,282 @@ +"""Rendering of a dataset's history as a timeline. + +This is the only layer that knows what a dataset's history looks like on +screen. It builds renderables and returns them; printing is the caller's job. + +The timeline is a two-column ``Table.grid``: a rail on the left, one glyph per +line, and the snapshot's facts on the right. Keeping one grid row per rendered +line is what holds the rail in step with the text beside it, and it is why the +content column must keep rich's default overflow: a column told not to wrap is +dropped entirely when the grid has to shrink, taking the rail's glyphs with it +and leaving a plausible-looking but railless timeline on narrow terminals. + +Every value that reaches a cell comes from the store, which means it comes from +whatever the user passed to `dstrack track --name` or from a path on disk. +Cells are therefore assembled as `Text`, never interpolated into rich's markup: +a dataset named ``sales[/]report`` is not markup to be parsed, and treating it +as such raises rather than prints. +""" + +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime +from typing import Final + +from rich.console import Group, RenderableType +from rich.table import Table +from rich.text import Text + +from dstrack.store import LogEntry + +_MISSING: Final = "\N{EM DASH}" +_NODE: Final = "\N{BLACK CIRCLE}" +_RAIL: Final = "\N{BOX DRAWINGS LIGHT VERTICAL}" +_SHORT_ID_LEN: Final = 8 + +# Largest unit first; each is the number of seconds in one of that unit. +_UNITS: Final[tuple[tuple[str, int], ...]] = ( + ("year", 365 * 24 * 3600), + ("month", 30 * 24 * 3600), + ("week", 7 * 24 * 3600), + ("day", 24 * 3600), + ("hour", 3600), + ("minute", 60), + ("second", 1), +) + + +def render_history( + entries: Sequence[LogEntry], + *, + parents: Mapping[str, LogEntry], + head_id: str, + now: datetime, + oneline: bool = False, +) -> RenderableType: + """Render a dataset's snapshots as a timeline. + + Args: + entries: The snapshots to show, in display order. May be a slice of + the dataset's full history. + parents: Each snapshot's predecessor, keyed by ``snapshot_id``, built + from the *full* history rather than `entries`. Passed separately + so that a snapshot whose parent was sliced off or reordered still + shows how it differed from it. + head_id: Id of the dataset's latest snapshot, marked in the output. + now: The moment to measure each snapshot's age against. + oneline: Condense each snapshot to a single line. + + Returns: + A renderable timeline, newest or oldest first according to the order + of `entries`. + """ + grid = Table.grid(padding=(0, 1)) + grid.add_column(no_wrap=True) + # No overflow or wrap settings: rich's defaults ellipsize each cell to one + # line, which is what keeps every content line paired with one rail glyph. + grid.add_column() + + for position, entry in enumerate(entries): + is_last = position == len(entries) - 1 + parent = parents.get(entry.snapshot_id) + lines = ( + [_oneline(entry, parent=parent, head_id=head_id, now=now)] + if oneline + else _detail_lines(entry, parent=parent, head_id=head_id, now=now) + ) + + for offset, line in enumerate(lines): + grid.add_row( + _glyph(entry, head_id=head_id, first=offset == 0, last=is_last), line + ) + if not oneline and not is_last: + grid.add_row(Text(_RAIL, style="dim cyan"), "") + + return Group(grid) + + +def _glyph(entry: LogEntry, *, head_id: str, first: bool, last: bool) -> Text: + """Return the rail character for one rendered line. + + Args: + entry: The snapshot the line belongs to. + head_id: Id of the dataset's latest snapshot. + first: Whether this is the snapshot's first line, which carries the + node rather than the rail. + last: Whether this is the oldest snapshot shown. Its continuation lines + carry no rail, so the timeline ends at the node instead of trailing + off below it. + + Returns: + The styled glyph to place in the rail column. + """ + if first: + style = "bold cyan" if entry.snapshot_id == head_id else "cyan" + return Text(_NODE, style=style) + return Text(" " if last else _RAIL, style="dim cyan") + + +def _detail_lines( + entry: LogEntry, *, parent: LogEntry | None, head_id: str, now: datetime +) -> list[Text]: + """Render one snapshot as its multi-line timeline entry. + + Args: + entry: The snapshot to render. + parent: The snapshot it succeeds, for computing deltas, or ``None``. + head_id: Id of the dataset's latest snapshot. + now: The moment to measure the snapshot's age against. + + Returns: + One `Text` per line, each rendered beside its own rail glyph. + """ + title = Text() + title.append(_short_id(entry.snapshot_id), style="bold yellow") + title.append(" ") + title.append(entry.dataset_name or _MISSING, style="bold") + if entry.snapshot_id == head_id: + title.append(" HEAD", style="bold cyan") + + when = Text() + when.append(_humanize(entry.created_at, now)) + when.append(" by ", style="dim") + when.append(entry.created_by or _MISSING) + + counts = Text() + counts.append_text( + _count(entry.num_rows, "row", parent.num_rows if parent else None) + ) + counts.append(" ") + counts.append_text( + _count(entry.num_columns, "col", parent.num_columns if parent else None) + ) + + return [title, when, counts, Text(entry.dataset_path or _MISSING, style="dim")] + + +def _oneline( + entry: LogEntry, *, parent: LogEntry | None, head_id: str, now: datetime +) -> Text: + """Render one snapshot as a single timeline line. + + Args: + entry: The snapshot to render. + parent: The snapshot it succeeds, for computing deltas, or ``None``. + head_id: Id of the dataset's latest snapshot. + now: The moment to measure the snapshot's age against. + + Returns: + The snapshot's identity, age, name and row count on one line. + """ + line = Text() + line.append(_short_id(entry.snapshot_id), style="bold yellow") + line.append(" ") + line.append(_humanize(entry.created_at, now), style="dim") + line.append(" ") + line.append(entry.dataset_name or _MISSING, style="bold") + line.append(" ") + line.append_text(_count(entry.num_rows, "row", parent.num_rows if parent else None)) + if entry.snapshot_id == head_id: + line.append(" HEAD", style="bold cyan") + return line + + +def _count(value: int | None, unit: str, parent: int | None) -> Text: + """Render a count and how it changed from its parent. + + Args: + value: The count, or ``None`` if the snapshot did not record one. + unit: Singular name of what is being counted, e.g. ``"row"``. + parent: The parent snapshot's count, or ``None`` if there is no parent + or it recorded none. + + Returns: + The count, or an em-dash if it is missing, followed by its signed + change when one can be computed. + """ + text = Text() + text.append(_MISSING if value is None else f"{value:,}", style="bold") + text.append(f" {unit}" if value == 1 else f" {unit}s", style="dim") + + delta = _delta(value, parent) + if delta is not None: + text.append(" ") + text.append(delta[0], style=delta[1]) + return text + + +def _delta(value: int | None, parent: int | None) -> tuple[str, str] | None: + """Return a count's signed change and the style to render it in. + + Args: + value: The snapshot's count. + parent: The parent snapshot's count. + + Returns: + The change and its style, or ``None`` when there is nothing to say: + no parent, an unrecorded count on either side, or no change at all. An + em-dash marks a missing *value*; a delta that cannot be computed is + simply not shown, rather than shown as unknown. + """ + if value is None or parent is None or value == parent: + return None + change = value - parent + return f"{change:+,}", "green" if change > 0 else "red" + + +def _short_id(snapshot_id: str) -> str: + """Return the leading characters of a snapshot id, as git shows a commit.""" + return snapshot_id[:_SHORT_ID_LEN] + + +def _humanize(created_at: str | None, now: datetime) -> str: + """Render when a snapshot was taken, relative to `now`. + + Args: + created_at: The recorded timestamp, as it appears in the log. + now: The moment to measure against. + + Returns: + An approximate age such as ``"3 days ago"``, or an em-dash if the + timestamp is missing or unreadable. A snapshot dated in the future + reads as ``"in 3 days"``: the store travels between machines by git, + so a teammate's clock can legitimately be ahead of this one. + """ + when = _parse_created_at(created_at) + if when is None: + return _MISSING + + seconds = int((now - when).total_seconds()) + magnitude = abs(seconds) + if magnitude < 1: + return "just now" + + for unit, size in _UNITS: + if magnitude >= size: + count = magnitude // size + plural = "" if count == 1 else "s" + return ( + f"{count} {unit}{plural} ago" + if seconds > 0 + else f"in {count} {unit}{plural}" + ) + return "just now" + + +def _parse_created_at(created_at: str | None) -> datetime | None: + """Parse a recorded timestamp into an aware datetime. + + Args: + created_at: The recorded timestamp, as it appears in the log. + + Returns: + The timestamp, or ``None`` if it is missing or unparsable. A timestamp + without an offset is read as UTC, which is what the store writes; it + cannot be left naive, since subtracting it from an aware `now` raises. + """ + if created_at is None: + return None + try: + parsed = datetime.fromisoformat(created_at) + except ValueError: + return None + return parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed diff --git a/src/dstrack/cache.py b/src/dstrack/cache.py new file mode 100644 index 0000000..547cda7 --- /dev/null +++ b/src/dstrack/cache.py @@ -0,0 +1,448 @@ +"""A SQLite index over the store's dataset logs. + +The index at ``.dstrack/.cache/index.db`` answers "what is this dataset's +history?" and "which dataset was recorded at this path?" without reparsing +every dataset's ``log.jsonl`` on every invocation. + +It is a cache, never a second source of truth. ``log.jsonl`` is committed to +git and is authoritative; the index is gitignored, derived from it, +and safe to delete at any moment. Two consequences shape this module: + +- **It can be stale, so every read resyncs first.** The index does not travel + with the repository, and `dstrack track` does not write to it, so a clone, a + pull, or a snapshot taken on another machine all leave it behind. Before any + query, [sync][dstrack.cache.sync] fingerprints each dataset's log and + reimports the ones that changed. When nothing changed, that costs one + ``stat`` and one tiny ``HEAD`` read per dataset. +- **It can be discarded, so damage is repaired rather than reported.** An + unreadable file or a schema from another version is rebuilt in place; there + is nothing to lose by doing so. + +``HEAD`` is deliberately *not* cached. It is a few bytes, it is the store's +commit pointer, and reading it fresh keeps +[write_snapshot][dstrack.store.write_snapshot]'s central invariant intact: the +index caches parsed lines, while what those lines *mean* is still decided by +the ``HEAD`` on disk. +""" + +import contextlib +import logging +import sqlite3 +from collections.abc import Iterator +from pathlib import Path +from typing import Final + +from dstrack import store +from dstrack.errors import IndexUnusable +from dstrack.store import DatasetSummary, LogEntry + +_log = logging.getLogger(__name__) + +CACHE_DIRNAME: Final = ".cache" +INDEX_FILENAME: Final = "index.db" + +# Bump whenever the schema below changes: a mismatch rebuilds the index rather +# than querying it with the wrong shape. +_SCHEMA_VERSION: Final = "1" + +_SCHEMA: Final = """ +CREATE TABLE meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); + +CREATE TABLE datasets ( + dataset_id TEXT PRIMARY KEY, + log_size INTEGER NOT NULL, + log_mtime_ns INTEGER NOT NULL, + head_snapshot_id TEXT +); + +CREATE TABLE snapshots ( + snapshot_id TEXT PRIMARY KEY, + dataset_id TEXT NOT NULL REFERENCES datasets(dataset_id) ON DELETE CASCADE, + parent_snapshot_id TEXT, + created_at TEXT, + created_by TEXT, + dataset_name TEXT, + dataset_path TEXT, + num_rows INTEGER, + num_columns INTEGER +); + +CREATE INDEX idx_snapshots_dataset ON snapshots(dataset_id); +""" + +_SNAPSHOT_COLUMNS: Final = ( + "snapshot_id", + "dataset_id", + "parent_snapshot_id", + "created_at", + "created_by", + "dataset_name", + "dataset_path", + "num_rows", + "num_columns", +) + + +def index_path(store_root: Path) -> Path: + """Return the path of the store's index. Need not exist. + + Args: + store_root: Path to the ``.dstrack/`` directory. + + Returns: + The index's path, inside the store's gitignored cache directory. + """ + return store_root / CACHE_DIRNAME / INDEX_FILENAME + + +def index_exists(store_root: Path) -> bool: + """Return whether the store's index has been built yet. + + Args: + store_root: Path to the ``.dstrack/`` directory. + + Returns: + ``True`` if an index file is present. It may still turn out to be + unusable, in which case [sync][dstrack.cache.sync] rebuilds it. + """ + return index_path(store_root).is_file() + + +def sync(store_root: Path) -> str | None: + """Bring the index in line with the store's logs. + + Reimports each dataset whose ``log.jsonl`` changed since it was last + indexed, and forgets datasets that are no longer on disk. Builds the index + if it is absent, and rebuilds it if it cannot be used as-is. + + Args: + store_root: Path to the ``.dstrack/`` directory, e.g. from + [resolve_store_root][dstrack.paths.resolve_store_root]. + + Returns: + ``None`` if the index was built from scratch or updated in place, or a + human-readable reason if an existing index had to be discarded and + rebuilt, for the caller to report. Discarding is recovery, not an + error: the index holds nothing that is not derived from the logs. + + Raises: + StoreCorruptionError: If a dataset's log holds a malformed entry. + OSError: If the store cannot be read, or the index cannot be written. + """ + reason: str | None = None + if index_exists(store_root): + try: + with _connect(store_root) as conn: + _check_schema_version(conn) + _sync(conn, store_root) + return None + except (sqlite3.DatabaseError, IndexUnusable) as e: + reason = str(e) + _log.warning(f"Rebuilding the snapshot index: {e}") + + _create_empty(store_root) + with _connect(store_root) as conn: + _sync(conn, store_root) + return reason + + +def query_history(dataset_id: str, *, store_root: Path) -> list[LogEntry]: + """Return a dataset's committed history, newest first. + + Walks back from the ``HEAD`` recorded by the last + [sync][dstrack.cache.sync], rather than re-reading ``HEAD`` from disk. + That is what keeps the answer self-consistent: a sync reads each dataset's + ``HEAD`` immediately before importing its log, so every snapshot that head + names is guaranteed to be indexed. Reading ``HEAD`` again afterwards could + pick up a snapshot a concurrent `dstrack track` committed since, whose log + line this index has not seen, which is indistinguishable from a log that + has lost an entry. + + Call [sync][dstrack.cache.sync] first, or the answer is as old as the last + one. + + Args: + dataset_id: Dataset whose history to read. + store_root: Path to the ``.dstrack/`` directory. + + Returns: + The dataset's snapshots, ordered newest first and ending at the one + with no parent. Empty if the dataset is not indexed, or has no + committed snapshot yet. + + Raises: + StoreCorruptionError: If the indexed lineage is missing an entry or + loops back on itself. + """ + with _connect(store_root) as conn: + head_row = conn.execute( + "SELECT head_snapshot_id FROM datasets WHERE dataset_id = ?", + (dataset_id,), + ).fetchone() + if head_row is None or head_row["head_snapshot_id"] is None: + return [] + head: str = head_row["head_snapshot_id"] + rows = conn.execute( + "SELECT * FROM snapshots WHERE dataset_id = ?", (dataset_id,) + ).fetchall() + index = {row["snapshot_id"]: _entry_from_row(row) for row in rows} + return store.walk_lineage(index, head) + + +def find_dataset_by_path(dataset_path: str, *, store_root: Path) -> str | None: + """Return the id of the dataset whose latest snapshot has this path. + + Matches the same way [write_snapshot][dstrack.store.write_snapshot] does: + against each dataset's ``HEAD`` snapshot only, never its older ones. A + dataset whose file has since been renamed therefore does not match the new + path, and must be named by its id instead. + + Args: + dataset_path: Relative POSIX path to match, computed against the same + path root the snapshot was recorded with. + store_root: Path to the ``.dstrack/`` directory. + + Returns: + The matching ``dataset_id``, or ``None`` if no dataset's latest + snapshot recorded this path. + """ + with _connect(store_root) as conn: + row = conn.execute( + "SELECT d.dataset_id FROM datasets d " + "JOIN snapshots s ON s.snapshot_id = d.head_snapshot_id " + "WHERE s.dataset_path = ? ORDER BY d.dataset_id LIMIT 1", + (dataset_path,), + ).fetchone() + if row is None: + return None + dataset_id: str = row["dataset_id"] + return dataset_id + + +def list_datasets(*, store_root: Path) -> list[DatasetSummary]: + """Return every indexed dataset, described by its latest snapshot. + + Args: + store_root: Path to the ``.dstrack/`` directory. + + Returns: + One summary per dataset, sorted by id. A dataset with no committed + snapshot yet is included, with a ``head`` of ``None``. + """ + with _connect(store_root) as conn: + rows = conn.execute( + "SELECT d.dataset_id AS id, s.* FROM datasets d " + "LEFT JOIN snapshots s ON s.snapshot_id = d.head_snapshot_id " + "ORDER BY d.dataset_id" + ).fetchall() + return [ + DatasetSummary( + dataset_id=row["id"], + head=_entry_from_row(row) if row["snapshot_id"] is not None else None, + ) + for row in rows + ] + + +@contextlib.contextmanager +def _connect(store_root: Path) -> Iterator[sqlite3.Connection]: + """Open the index, creating its parent directory if needed. + + Args: + store_root: Path to the ``.dstrack/`` directory. + + Yields: + A connection with row access by name and foreign keys enforced, so a + dataset's rows are cleared with it. Uncommitted work is rolled back if + the block raises. + """ + path = index_path(store_root) + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(path) + try: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + yield conn + finally: + conn.close() + + +def _create_empty(store_root: Path) -> None: + """Replace the index with an empty one at the current schema version. + + Args: + store_root: Path to the ``.dstrack/`` directory. + + Raises: + OSError: If the old index cannot be removed or the new one written. + """ + index_path(store_root).unlink(missing_ok=True) + with _connect(store_root) as conn: + conn.executescript(_SCHEMA) + conn.execute( + "INSERT INTO meta (key, value) VALUES ('schema_version', ?)", + (_SCHEMA_VERSION,), + ) + conn.commit() + + +def _check_schema_version(conn: sqlite3.Connection) -> None: + """Raise unless the index was built by this version of the schema. + + Args: + conn: A connection to the index. + + Raises: + _IndexUnusable: If the index records a different schema version, or no + version at all. + """ + try: + row = conn.execute( + "SELECT value FROM meta WHERE key = 'schema_version'" + ).fetchone() + except sqlite3.DatabaseError as e: + raise IndexUnusable(f"its schema could not be read ({e})") from e + if row is None: + raise IndexUnusable("it records no schema version") + if row["value"] != _SCHEMA_VERSION: + raise IndexUnusable( + f"it was built for schema version {row['value']}, " + f"but this dstrack expects {_SCHEMA_VERSION}" + ) + + +def _sync(conn: sqlite3.Connection, store_root: Path) -> None: + """Reimport every dataset whose log changed, and forget deleted ones. + + Runs as a single transaction, so an interrupted sync leaves the index at + its previous state rather than partly reimported. + + Args: + conn: A connection to the index. + store_root: Path to the ``.dstrack/`` directory. + + Raises: + StoreCorruptionError: If a dataset's log holds a malformed entry. + OSError: If the store cannot be read. + """ + fingerprints = { + row["dataset_id"]: (row["log_size"], row["log_mtime_ns"]) + for row in conn.execute( + "SELECT dataset_id, log_size, log_mtime_ns FROM datasets" + ) + } + + on_disk = store.list_dataset_ids(store_root=store_root) + for dataset_id in on_disk: + # HEAD before the log, so a concurrent `track` cannot move HEAD onto a + # snapshot whose line this sync has already read past. + head = store.read_head(dataset_id, store_root=store_root) + fingerprint = _fingerprint(dataset_id, store_root=store_root) + + conn.execute( + "INSERT INTO datasets (dataset_id, log_size, log_mtime_ns, head_snapshot_id) " + "VALUES (?, ?, ?, ?) ON CONFLICT(dataset_id) DO UPDATE SET " + "log_size = excluded.log_size, " + "log_mtime_ns = excluded.log_mtime_ns, " + "head_snapshot_id = excluded.head_snapshot_id", + (dataset_id, fingerprint[0], fingerprint[1], head), + ) + + if fingerprints.get(dataset_id) == fingerprint: + continue + _reimport(conn, dataset_id, store_root=store_root) + + for dataset_id in fingerprints.keys() - set(on_disk): + conn.execute("DELETE FROM datasets WHERE dataset_id = ?", (dataset_id,)) + + conn.commit() + + +def _reimport(conn: sqlite3.Connection, dataset_id: str, *, store_root: Path) -> None: + """Replace a dataset's indexed rows with what its log currently says. + + The whole log is reparsed rather than appended from a recorded offset. + ``log.jsonl`` is append-only in normal operation, but the store's own + corruption messages instruct users to delete an offending line by hand, so + an offset cannot be trusted to still point at the same entry. The logs are + small by design, which is what makes reparsing the cheaper bet. + + Args: + conn: A connection to the index, inside a transaction. + dataset_id: Dataset to reimport. + store_root: Path to the ``.dstrack/`` directory. + + Raises: + StoreCorruptionError: If the log holds a malformed entry. + OSError: If the log cannot be read. + """ + conn.execute("DELETE FROM snapshots WHERE dataset_id = ?", (dataset_id,)) + entries = store.read_log_entries(dataset_id, store_root=store_root) + placeholders = ", ".join("?" * len(_SNAPSHOT_COLUMNS)) + # A duplicated line is ignored rather than fatal: the first occurrence + # wins, matching how the lineage walk would have reached it. + conn.executemany( + f"INSERT OR IGNORE INTO snapshots ({', '.join(_SNAPSHOT_COLUMNS)}) " + f"VALUES ({placeholders})", + [ + ( + entry.snapshot_id, + dataset_id, + entry.parent_snapshot_id, + entry.created_at, + entry.created_by, + entry.dataset_name, + entry.dataset_path, + entry.num_rows, + entry.num_columns, + ) + for entry in entries + ], + ) + + +def _fingerprint(dataset_id: str, *, store_root: Path) -> tuple[int, int]: + """Return a cheap signature of a dataset's log, to detect changes. + + Size and modification time, the same trade-off git's index makes: it + misses an edit that preserves both, which a rebuild is the answer to, and + costs one ``stat`` rather than a reparse when nothing changed. + + Args: + dataset_id: Dataset whose log to fingerprint. + store_root: Path to the ``.dstrack/`` directory. + + Returns: + The log's size and modification time in nanoseconds, or zeroes if the + dataset has no log yet. + """ + path = store.dataset_log_path(dataset_id, store_root=store_root) + try: + stat = path.stat() + except FileNotFoundError: + return (0, 0) + return (stat.st_size, stat.st_mtime_ns) + + +def _entry_from_row(row: sqlite3.Row) -> LogEntry: + """Build a [LogEntry][dstrack.store.LogEntry] from an indexed row. + + Args: + row: A ``snapshots`` row, or a join that selects its columns. + + Returns: + The row as a [LogEntry][dstrack.store.LogEntry]. + """ + return LogEntry( + snapshot_id=row["snapshot_id"], + parent_snapshot_id=row["parent_snapshot_id"], + created_at=row["created_at"], + created_by=row["created_by"], + dataset_name=row["dataset_name"], + dataset_path=row["dataset_path"], + num_rows=row["num_rows"], + num_columns=row["num_columns"], + ) diff --git a/src/dstrack/console.py b/src/dstrack/console.py index 0fe73ad..c19eec2 100644 --- a/src/dstrack/console.py +++ b/src/dstrack/console.py @@ -2,28 +2,46 @@ Gives every command a consistent way to report results to the user, independent of the logging configuration. + +Messages are plain text, not rich markup. What they report is routinely +user-derived -- a dataset name from `dstrack track --name`, a path, an +exception carrying either -- and square brackets are legal in all of them. +Rendering such a message as markup would swallow ``[bold]`` as styling and +raise outright on ``[/]``, so each helper escapes its message and applies +styling only to the icon it puts in front. """ -from rich.console import Console +from rich.console import Console, RenderableType +from rich.markup import escape console = Console() +def display(renderable: RenderableType) -> None: + """Print a rich renderable, e.g. a table or a timeline. + + For command output that is built rather than phrased: unlike the message + helpers below, what is passed is rendered as-is, so a caller assembling + text from user-derived values should build it with `rich.text.Text`. + """ + console.print(renderable) + + def success(message: str) -> None: """Print a success message prefixed with a check mark.""" - console.print(f"[bold green]\N{HEAVY CHECK MARK}[/bold green] {message}") + console.print(f"[bold green]\N{HEAVY CHECK MARK}[/bold green] {escape(message)}") def warning(message: str) -> None: """Print a warning message prefixed with a lightning bolt.""" - console.print(f"[bold yellow]\N{HIGH VOLTAGE SIGN}[/bold yellow] {message}") + console.print(f"[bold yellow]\N{HIGH VOLTAGE SIGN}[/bold yellow] {escape(message)}") def error(message: str) -> None: """Print an error message prefixed with a bug icon.""" - console.print(f"[bold red]\N{BUG}[/bold red] {message}") + console.print(f"[bold red]\N{BUG}[/bold red] {escape(message)}") def info(message: str) -> None: """Print general information with an info icon.""" - console.print(f"[bold blue]\N{INFORMATION SOURCE}[/bold blue] {message}") + console.print(f"[bold blue]\N{INFORMATION SOURCE}[/bold blue] {escape(message)}") diff --git a/src/dstrack/errors.py b/src/dstrack/errors.py index aadda49..c6e493b 100644 --- a/src/dstrack/errors.py +++ b/src/dstrack/errors.py @@ -16,3 +16,7 @@ class DatasetNotFoundError(Exception): class InputTooLargeError(Exception): """Raised when a dataset exceeds the configured snapshot row limit.""" + + +class IndexUnusable(Exception): + """The index exists but cannot be queried, and must be rebuilt.""" diff --git a/src/dstrack/store.py b/src/dstrack/store.py index 0639dfa..42f5cad 100644 --- a/src/dstrack/store.py +++ b/src/dstrack/store.py @@ -1,4 +1,4 @@ -"""Persistence of snapshots into the local store. +"""Persistence of snapshots into the local store, and reads back out of it. A snapshot is written under ``datasets//`` as its full JSON payload, a one-line append to ``log.jsonl``, and an updated ``HEAD``. The @@ -8,13 +8,23 @@ dataset's lineage rather than creating a new one. The three writes are not a single atomic transaction, so two safeguards keep -readers consistent. A store-wide advisory lock (:func:`_store_lock`) serializes +readers consistent. A store-wide advisory lock serializes the whole resolve-parent-then-write sequence, so concurrent writers cannot read the same ``HEAD`` as their parent nor mint two datasets for one path. And ``HEAD`` -- written last -- is the single source of truth for what is committed: recovery and path matching read the log entry that ``HEAD`` names rather than trusting the final line, so a ``log.jsonl`` left one entry ahead by a crash between the append and the ``HEAD`` write is ignored. + +Reading history back is therefore *reachability-based*, not positional. +[read_log_entries][dstrack.store.read_log_entries] returns every line a dataset's log holds, including +entries no ``HEAD`` ever named; [walk_lineage][dstrack.store.walk_lineage] then follows +``parent_snapshot_id`` links back from ``HEAD`` to select the ones that are +actually committed. Neither "read every line" nor "read every line up to +``HEAD``" is correct: a crash between the append and the ``HEAD`` write leaves +an uncommitted entry behind, and the *next* successful write appends after it, +stranding it in the middle of the file rather than at the end. Only +reachability from ``HEAD`` tells the two apart. """ import contextlib @@ -23,7 +33,7 @@ import sys import tempfile import uuid -from collections.abc import Iterator +from collections.abc import Iterator, Mapping from dataclasses import dataclass from pathlib import Path from typing import Any, Final @@ -43,6 +53,65 @@ "num_columns", ) +_DATASETS_DIRNAME: Final = "datasets" +_LOG_FILENAME: Final = "log.jsonl" +_HEAD_FILENAME: Final = "HEAD" +_SNAPSHOTS_DIRNAME: Final = "snapshots" + + +@dataclass(frozen=True) +class LogEntry: + """One snapshot as recorded in a dataset's ``log.jsonl``. + + The lightweight per-snapshot record, mirroring + log fields rather than the full + ``snapshots/.json`` payload. + + Every field except ``snapshot_id`` is optional. Log lines are written with + ``payload.get(key)`` (see [write_snapshot][dstrack.store.write_snapshot]), + so a snapshot built without a field records it as JSON null, and a line + written by a different version of dstrack may omit it entirely. + + Attributes: + snapshot_id: Identifier of the snapshot. Always present: an entry is + only ever reached via ``HEAD`` or a ``parent_snapshot_id`` link, + both of which name it. + parent_snapshot_id: Snapshot this one succeeds, or ``None`` for the + dataset's first snapshot. + created_at: ISO-8601 UTC timestamp, kept as the raw recorded string. + Deliberately not parsed here: an unparsable value is a display + concern, not a reason to fail a read. + created_by: Recorded author, or ``None`` if not recorded. + dataset_name: Human-readable name at snapshot time, or ``None``. + dataset_path: POSIX path relative to the path root in force at + snapshot time, or ``None``. + num_rows: Row count, or ``None`` if not recorded. + num_columns: Column count, or ``None`` if not recorded. + """ + + snapshot_id: str + parent_snapshot_id: str | None + created_at: str | None + created_by: str | None + dataset_name: str | None + dataset_path: str | None + num_rows: int | None + num_columns: int | None + + +@dataclass(frozen=True) +class DatasetSummary: + """A dataset in the store, described by its latest committed snapshot. + + Attributes: + dataset_id: The dataset's directory name under ``datasets/``. + head: The log entry the dataset's ``HEAD`` names, or ``None`` if the + dataset has no committed snapshot yet. + """ + + dataset_id: str + head: LogEntry | None + @dataclass(frozen=True) class SnapshotWriteResult: @@ -106,7 +175,7 @@ def write_snapshot( line that is not valid JSON. OSError: If the store cannot be written to. """ - datasets_dir = store_root / "datasets" + datasets_dir = _datasets_dir(store_root) dataset_path = snapshot["dataset_path"] snapshot_id = snapshot["snapshot_id"] @@ -129,7 +198,7 @@ def write_snapshot( dataset_id = str(uuid.uuid4()) dataset_dir = datasets_dir / dataset_id - snapshots_dir = dataset_dir / "snapshots" + snapshots_dir = dataset_dir / _SNAPSHOTS_DIRNAME snapshot_path = snapshots_dir / f"{snapshot_id}.json" # snapshot_id arrives in the payload from outside, so confirm the file # lands inside the dataset's snapshots/ before creating any directory. @@ -148,10 +217,10 @@ def write_snapshot( _atomic_write(snapshot_path, json.dumps(payload, indent=2) + "\n") log_line = {key: payload.get(key) for key in _LOG_FIELDS} - with (dataset_dir / "log.jsonl").open("a", encoding="utf-8") as fh: + with (dataset_dir / _LOG_FILENAME).open("a", encoding="utf-8") as fh: fh.write(json.dumps(log_line) + "\n") - _atomic_write(dataset_dir / "HEAD", snapshot_id + "\n") + _atomic_write(dataset_dir / _HEAD_FILENAME, snapshot_id + "\n") return SnapshotWriteResult( dataset_id=dataset_id, @@ -162,6 +231,319 @@ def write_snapshot( ) +def list_dataset_ids(*, store_root: Path) -> list[str]: + """Return the id of every dataset in the store, sorted. + + Args: + store_root: Path to the ``.dstrack/`` directory, e.g. from + [resolve_store_root][dstrack.paths.resolve_store_root]. + + Returns: + Each dataset's id, i.e. its directory name under ``datasets/``. Empty + if the store holds no datasets yet. + """ + return [d.name for d in _iter_dataset_dirs(_datasets_dir(store_root))] + + +def dataset_exists(dataset_id: str, *, store_root: Path) -> bool: + """Return whether the store holds a dataset with this id. + + Args: + dataset_id: The dataset id, which may come from the user. + store_root: Path to the ``.dstrack/`` directory. + + Returns: + ``True`` if the dataset has a directory in the store, whether or not + it has been snapshotted yet. + + Raises: + ValueError: If ``dataset_id`` resolves to a path outside the store. + """ + return _resolve_dataset_dir(dataset_id, store_root).is_dir() + + +def read_head(dataset_id: str, *, store_root: Path) -> str | None: + """Return the id of the snapshot a dataset's ``HEAD`` names. + + ``HEAD`` is the single source of truth for what a dataset has committed, + so callers that derive history from it should read it *before* reading + ``log.jsonl``: a snapshot's log line is appended before the ``HEAD`` write + that names it, so any ``HEAD`` observed at a given moment is guaranteed to + have its whole ancestry already on disk. Reading them the other way round + can observe a ``HEAD`` naming a snapshot whose line was appended after the + log was read, which looks indistinguishable from corruption. + + Args: + dataset_id: Dataset whose ``HEAD`` to read. + store_root: Path to the ``.dstrack/`` directory. + + Returns: + The latest committed snapshot's id, or ``None`` if the dataset does + not exist or has never been snapshotted. + + Raises: + ValueError: If ``dataset_id`` resolves to a path outside the store. + """ + return _read_head(_resolve_dataset_dir(dataset_id, store_root)) + + +def dataset_log_path(dataset_id: str, *, store_root: Path) -> Path: + """Return the path of a dataset's ``log.jsonl``. Need not exist. + + Args: + dataset_id: Dataset whose log to locate. + store_root: Path to the ``.dstrack/`` directory. + + Returns: + The path of the dataset's append-only log. + + Raises: + ValueError: If ``dataset_id`` resolves to a path outside the store. + """ + return _resolve_dataset_dir(dataset_id, store_root) / _LOG_FILENAME + + +def read_log_entries(dataset_id: str, *, store_root: Path) -> list[LogEntry]: + """Return every entry in a dataset's ``log.jsonl``, in file order. + + Includes entries that no ``HEAD`` ever named, so the result is what the + log *says*, not what the dataset has *committed*. Pass the result through + [walk_lineage][dstrack.store.walk_lineage] to select the committed + lineage. + + A trailing line that is not valid JSON and is not newline-terminated is + treated as an interrupted append and dropped, not reported as corruption: + the log is appended to before ``HEAD`` moves, so a torn final line is a + snapshot that was never committed and is expected after a crash. A + malformed line anywhere else was written completely and is corruption. + + Args: + dataset_id: Dataset whose log to read. + store_root: Path to the ``.dstrack/`` directory. + + Returns: + One entry per log line, oldest first. Empty if the dataset does not + exist or has no log yet. + + Raises: + ValueError: If ``dataset_id`` resolves to a path outside the store. + StoreCorruptionError: If a complete log line is not a valid JSON + object, or records a ``snapshot_id`` or ``parent_snapshot_id`` + that is not a string. + OSError: If the log cannot be read. + """ + path = dataset_log_path(dataset_id, store_root=store_root) + if not path.is_file(): + return [] + + lines = path.read_text(encoding="utf-8").splitlines(keepends=True) + entries: list[LogEntry] = [] + for index, line in enumerate(lines): + if not line.strip(): + continue + try: + data = json.loads(line) + except json.JSONDecodeError as e: + is_torn_append = index == len(lines) - 1 and not line.endswith("\n") + if is_torn_append: + break + raise StoreCorruptionError( + f"An entry of `{path}` is not valid JSON. The file may have " + "been truncated by an interrupted write; restore it from git " + f"or delete the offending line. Error found in line loc {index}." + ) from e + if not isinstance(data, dict): + raise StoreCorruptionError( + f"An entry of `{path}` is not a JSON object. Restore the file " + "from git or delete the offending line." + ) + entries.append(_log_entry_from_line(data, path=path)) + return entries + + +def walk_lineage(index: Mapping[str, LogEntry], head: str) -> list[LogEntry]: + """Select the snapshots reachable from ``head``, newest first. + + Follows ``parent_snapshot_id`` links back from ``head``, which is what + makes a history *committed* rather than merely *recorded*. Entries the + walk does not reach are ignored: a crash between a log append and the + ``HEAD`` write (see [write_snapshot][dstrack.store.write_snapshot]) leaves + an entry the store never committed, and a later successful write appends + after it, stranding it mid-file rather than at the end. Reachability is + the only thing that tells such an entry apart from a real snapshot. + + Args: + index: Every entry the dataset's log holds, keyed by ``snapshot_id``, + e.g. built from [read_log_entries][dstrack.store.read_log_entries]. + head: Id of the snapshot to walk back from, i.e. the dataset's + ``HEAD``. + + Returns: + The lineage ``head`` names, ordered newest first and ending at the + snapshot with no parent. + + Raises: + StoreCorruptionError: If the lineage names a snapshot ``index`` has no + entry for, or if the ``parent_snapshot_id`` links form a cycle. + """ + lineage: list[LogEntry] = [] + seen: set[str] = set() + current: str | None = head + child: str | None = None + + while current is not None: + if current in seen: + raise StoreCorruptionError( + f"The history of snapshot {head!r} loops back on itself at " + f"{current!r}. A snapshot cannot be its own ancestor; the log " + "has been edited into a cycle. Restore it from git." + ) + seen.add(current) + + entry = index.get(current) + if entry is None: + missing = ( + f"The log has no entry for snapshot {current!r}" + if child is None + else f"The log has no entry for snapshot {current!r}, named as " + f"the parent of {child!r}" + ) + raise StoreCorruptionError( + f"{missing}. Part of the dataset's lineage is missing; restore " + "the log from git." + ) + + lineage.append(entry) + child = current + current = entry.parent_snapshot_id + + return lineage + + +def _log_entry_from_line(line: dict[str, Any], *, path: Path) -> LogEntry: + """Build a [LogEntry][dstrack.store.LogEntry] from a parsed log line. + + Reads each field by name and ignores any it does not know, rather than + unpacking the line. A store written by a newer dstrack would make ``LogEntry(**line)`` raise, and + ``.dstrack/`` is committed to git, so such a store legitimately reaches an + older client by way of a clone or a pull. + + Fields are validated by the job they do. ``snapshot_id`` and + ``parent_snapshot_id`` are structural: they drive + [walk_lineage][dstrack.store.walk_lineage], so a wrong type there is + corruption. The rest are cosmetic, and a wrong type is coerced to ``None`` + for the caller to render as missing: the store's own corruption messages + tell users to hand-edit the log, so hand-edited damage is expected, and a + display command must not die on a field it only prints. + + Args: + line: A parsed ``log.jsonl`` line. + path: The log the line came from, used in error messages. + + Returns: + The line as a [LogEntry][dstrack.store.LogEntry]. + + Raises: + StoreCorruptionError: If ``snapshot_id`` is missing, empty, or not a + string, or if ``parent_snapshot_id`` is present but not a string. + """ + snapshot_id = line.get("snapshot_id") + if not isinstance(snapshot_id, str) or not snapshot_id: + raise StoreCorruptionError( + f"An entry of `{path}` has no usable `snapshot_id`. Every entry " + "must identify its snapshot; restore the file from git or delete " + "the offending line." + ) + + parent_snapshot_id = line.get("parent_snapshot_id") + if parent_snapshot_id is not None and not isinstance(parent_snapshot_id, str): + raise StoreCorruptionError( + f"Entry {snapshot_id!r} of `{path}` records a " + "`parent_snapshot_id` that is not a snapshot id. Restore the file " + "from git or delete the offending line." + ) + + return LogEntry( + snapshot_id=snapshot_id, + parent_snapshot_id=parent_snapshot_id, + created_at=_optional_str(line.get("created_at")), + created_by=_optional_str(line.get("created_by")), + dataset_name=_optional_str(line.get("dataset_name")), + dataset_path=_optional_str(line.get("dataset_path")), + num_rows=_optional_int(line.get("num_rows")), + num_columns=_optional_int(line.get("num_columns")), + ) + + +def _optional_str(value: Any) -> str | None: + """Return `value` if it is a string, else ``None``.""" + return value if isinstance(value, str) else None + + +def _optional_int(value: Any) -> int | None: + """Return `value` if it is an integer, else ``None``. + + ``bool`` is a subclass of ``int``, so it is rejected explicitly: a count + recorded as ``true`` is damage, not the number one. + """ + return value if isinstance(value, int) and not isinstance(value, bool) else None + + +def _datasets_dir(store_root: Path) -> Path: + """Return the store's ``datasets/`` directory. Need not exist. + + Args: + store_root: Path to the ``.dstrack/`` directory. + + Returns: + The directory every dataset lives directly inside. + """ + return store_root / _DATASETS_DIRNAME + + +def _iter_dataset_dirs(datasets_dir: Path) -> Iterator[Path]: + """Yield each dataset's directory, in a stable order. + + Args: + datasets_dir: The store's ``datasets/`` directory. Need not exist. + + Yields: + Every directory directly inside ``datasets_dir``, sorted by name. + Non-directory entries are skipped, so a stray file in the store does + not read as a dataset. + """ + if not datasets_dir.is_dir(): + return + for dataset_dir in sorted(datasets_dir.iterdir()): + if dataset_dir.is_dir(): + yield dataset_dir + + +def _resolve_dataset_dir(dataset_id: str, store_root: Path) -> Path: + """Return a dataset's directory, rejecting an id that escapes the store. + + Args: + dataset_id: The dataset id, which may come from the user. + store_root: Path to the ``.dstrack/`` directory. + + Returns: + The ``datasets//`` directory. Need not exist. + + Raises: + ValueError: If ``dataset_id`` does not name a direct child of + ``datasets/``. + """ + datasets_dir = _datasets_dir(store_root) + dataset_dir = datasets_dir / dataset_id + _ensure_direct_child( + base=datasets_dir, + candidate=dataset_dir, + kind="dataset_id", + value=dataset_id, + ) + return dataset_dir + + def _ensure_direct_child(base: Path, candidate: Path, kind: str, value: str) -> None: """Raise unless ``candidate`` resolves to a direct child of ``base``. @@ -232,11 +614,7 @@ def _match_dataset_by_path(datasets_dir: Path, dataset_path: str) -> str | None: StoreCorruptionError: If a dataset's ``log.jsonl`` ends in a line that is not valid JSON. """ - if not datasets_dir.is_dir(): - return None - for dataset_dir in sorted(datasets_dir.iterdir()): - if not dataset_dir.is_dir(): - continue + for dataset_dir in _iter_dataset_dirs(datasets_dir): entry = _read_committed_log_entry(dataset_dir) if entry is not None and entry.get("dataset_path") == dataset_path: return dataset_dir.name @@ -267,7 +645,7 @@ def _read_committed_log_entry(dataset_dir: Path) -> dict[str, Any] | None: head = _read_head(dataset_dir) if head is None: return None - log_path = dataset_dir / "log.jsonl" + log_path = dataset_dir / _LOG_FILENAME if not log_path.is_file(): return None with log_path.open(encoding="utf-8") as fh: @@ -300,7 +678,7 @@ def _read_head(dataset_dir: Path) -> str | None: The latest snapshot's id, or ``None`` if the dataset has no ``HEAD`` yet, i.e. it has never been snapshotted. """ - head_path = dataset_dir / "HEAD" + head_path = dataset_dir / _HEAD_FILENAME if not head_path.is_file(): return None head = head_path.read_text(encoding="utf-8").strip() diff --git a/tests/test_log.py b/tests/test_log.py new file mode 100644 index 0000000..af3726a --- /dev/null +++ b/tests/test_log.py @@ -0,0 +1,604 @@ +"""Tests for the `dstrack log` command in src/dstrack/_log.py.""" + +import json +import shutil +import uuid +from collections.abc import Callable +from datetime import UTC, datetime, timedelta +from pathlib import Path +from types import SimpleNamespace + +import pytest +import typer +from typer.testing import CliRunner + +from dstrack import _log, cache +from dstrack._cli import app +from dstrack._log import ( + _canonical_uuid, + _ensure_index, + _known_datasets, + _resolve_target, +) +from dstrack._log_render import _humanize +from dstrack.errors import DatasetNotFoundError, StoreCorruptionError + +runner = CliRunner() + + +# --------------------------------------------------------------------------- +# Fixtures / helpers +# --------------------------------------------------------------------------- + + +@pytest.fixture +def store_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Create an empty local store under a fresh cwd and return its root. + + `log` resolves the store by walking up from the current directory, so the + tests run from inside `tmp_path`. `DSTRACK_ROOT_PATH` is cleared so an + ambient value cannot redirect the store elsewhere. + """ + monkeypatch.delenv("DSTRACK_ROOT_PATH", raising=False) + monkeypatch.chdir(tmp_path) + root = tmp_path / ".dstrack" + (root / "datasets").mkdir(parents=True) + return root + + +def _write_csv(path: Path, rows: int = 3) -> Path: + """Write a small, valid CSV file and return its path.""" + lines = ["a,b"] + [f"{i},{i * 2}" for i in range(rows)] + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + return path + + +def _track(root: Path, path: Path, *args: str) -> str: + """Track `path` through the CLI and return its dataset id.""" + result = runner.invoke(app, ["track", str(path), *args]) + assert result.exit_code == 0, result.output + return _dataset_id_of(root, path) + + +def _dataset_id_of(root: Path, path: Path) -> str: + """Return the id of the dataset whose snapshots recorded `path`.""" + for snapshot in root.glob("datasets/*/snapshots/*.json"): + payload = json.loads(snapshot.read_text(encoding="utf-8")) + if Path(payload["dataset_path"]).name == path.name: + return str(snapshot.parent.parent.name) + raise AssertionError(f"no dataset recorded for {path}") + + +def _pretend_terminal(monkeypatch: pytest.MonkeyPatch) -> None: + """Make `_log` see a terminal on stdin, so it offers to build the index. + + `CliRunner` installs its own `sys.stdin` for the duration of an invocation, + which would undo a patch of the real one; the module's own reference to + `sys` is replaced instead. Only `stdin.isatty` is ever read through it. + """ + monkeypatch.setattr( + _log, "sys", SimpleNamespace(stdin=SimpleNamespace(isatty=lambda: True)) + ) + + +def _log_cmd(*args: str) -> object: + """Invoke `dstrack log` with `--yes`, so the index is built unprompted.""" + return runner.invoke(app, ["log", *args, "--yes"]) + + +# --------------------------------------------------------------------------- +# Happy path +# --------------------------------------------------------------------------- + + +def test_log_by_path_shows_the_snapshot(tmp_path: Path, store_root: Path) -> None: + """A tracked file's path resolves to its dataset and shows its history.""" + csv = _write_csv(tmp_path / "data.csv") + _track(store_root, csv) + + result = _log_cmd(str(csv)) + + assert result.exit_code == 0, result.output + assert "data" in result.output + assert "3 rows" in result.output + assert "HEAD" in result.output + + +def test_log_by_dataset_id_shows_the_snapshot(tmp_path: Path, store_root: Path) -> None: + """A dataset id resolves without touching the filesystem path at all.""" + csv = _write_csv(tmp_path / "data.csv") + dataset_id = _track(store_root, csv) + csv.unlink() # The id must work even once the file is gone. + + result = _log_cmd(dataset_id) + + assert result.exit_code == 0, result.output + assert "3 rows" in result.output + + +def test_log_shows_newest_first(tmp_path: Path, store_root: Path) -> None: + """Without --reverse the latest snapshot is marked HEAD and comes first.""" + csv = _write_csv(tmp_path / "data.csv", rows=3) + _track(store_root, csv) + _write_csv(csv, rows=5) + _track(store_root, csv) + + result = _log_cmd(str(csv)) + + assert result.exit_code == 0, result.output + head_line = next(line for line in result.output.splitlines() if "HEAD" in line) + assert result.output.splitlines().index(head_line) < 4 + # The newest snapshot gained two rows over its parent. + assert "+2" in result.output + + +def test_log_reverse_shows_oldest_first(tmp_path: Path, store_root: Path) -> None: + """--reverse flips display order but keeps HEAD marked on the latest.""" + csv = _write_csv(tmp_path / "data.csv", rows=3) + _track(store_root, csv) + _write_csv(csv, rows=5) + _track(store_root, csv) + + result = _log_cmd(str(csv), "--reverse") + + assert result.exit_code == 0, result.output + lines = result.output.splitlines() + head_index = next(i for i, line in enumerate(lines) if "HEAD" in line) + assert head_index > 3, result.output + + +def test_log_limit_counts_back_from_head(tmp_path: Path, store_root: Path) -> None: + """--limit keeps the newest N snapshots, dropping the older ones.""" + csv = _write_csv(tmp_path / "data.csv", rows=3) + _track(store_root, csv) + _write_csv(csv, rows=5) + _track(store_root, csv) + dataset_id = _dataset_id_of(store_root, csv) + cache.sync(store_root) + history = cache.query_history(dataset_id, store_root=store_root) + + result = _log_cmd(str(csv), "-n", "1") + + assert result.exit_code == 0, result.output + assert history[0].snapshot_id[:8] in result.output + assert history[1].snapshot_id[:8] not in result.output + + +def test_log_limit_and_reverse_compose(tmp_path: Path, store_root: Path) -> None: + """--limit selects from HEAD first; --reverse only reorders what remains.""" + csv = _write_csv(tmp_path / "data.csv", rows=3) + _track(store_root, csv) + _write_csv(csv, rows=5) + _track(store_root, csv) + dataset_id = _dataset_id_of(store_root, csv) + cache.sync(store_root) + history = cache.query_history(dataset_id, store_root=store_root) + + result = _log_cmd(str(csv), "-n", "1", "--reverse") + + assert result.exit_code == 0, result.output + assert history[0].snapshot_id[:8] in result.output + assert history[1].snapshot_id[:8] not in result.output + + +def test_log_limit_below_one_is_rejected(tmp_path: Path, store_root: Path) -> None: + """--limit is constrained to at least one snapshot.""" + csv = _write_csv(tmp_path / "data.csv") + _track(store_root, csv) + + result = _log_cmd(str(csv), "-n", "0") + + assert result.exit_code == 2, result.output + + +def test_log_oneline_condenses_each_snapshot(tmp_path: Path, store_root: Path) -> None: + """--oneline renders one line per snapshot instead of a detail block.""" + csv = _write_csv(tmp_path / "data.csv") + _track(store_root, csv) + + detailed = _log_cmd(str(csv)) + condensed = _log_cmd(str(csv), "--oneline") + + assert condensed.exit_code == 0, condensed.output + assert len(condensed.output.splitlines()) < len(detailed.output.splitlines()) + assert "3 rows" in condensed.output + # The path is a detail-view line only. + assert "data.csv" not in condensed.output + + +def test_log_root_option_changes_the_path_root( + tmp_path: Path, store_root: Path +) -> None: + """--root matches the path root `track` recorded the dataset with.""" + nested = tmp_path / "nested" + nested.mkdir() + csv = _write_csv(nested / "data.csv") + _track(store_root, csv, "--root", str(nested)) + + without_root = _log_cmd(str(csv)) + with_root = _log_cmd(str(csv), "--root", str(nested)) + + assert without_root.exit_code == 1, without_root.output + assert with_root.exit_code == 0, with_root.output + + +# --------------------------------------------------------------------------- +# Failure paths +# --------------------------------------------------------------------------- + + +def test_log_without_a_store_exits_one( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """With no `.dstrack/` anywhere above the cwd there is no history to show.""" + monkeypatch.delenv("DSTRACK_ROOT_PATH", raising=False) + workdir = tmp_path / "workdir" + workdir.mkdir() + monkeypatch.chdir(workdir) + + result = _log_cmd("whatever") + + assert result.exit_code == 1, result.output + + +def test_log_unknown_dataset_id_lists_known_datasets( + tmp_path: Path, store_root: Path +) -> None: + """An id no dataset goes by fails, and the error names the ids that exist.""" + csv = _write_csv(tmp_path / "data.csv") + known = _track(store_root, csv) + + result = _log_cmd(str(uuid.uuid4())) + + assert result.exit_code == 1, result.output + assert "No dataset" in result.output + assert known[:8] in result.output + + +def test_log_untracked_path_explains_the_recorded_path( + tmp_path: Path, store_root: Path +) -> None: + """A path no dataset recorded reports what it was looked up as.""" + _write_csv(tmp_path / "data.csv") + + result = _log_cmd(str(tmp_path / "data.csv")) + + assert result.exit_code == 1, result.output + assert "No tracked dataset" in result.output + assert "Known datasets" in result.output + + +def test_log_dataset_without_snapshots_warns(tmp_path: Path, store_root: Path) -> None: + """A dataset directory with no committed snapshot has nothing to show.""" + dataset_id = str(uuid.uuid4()) + (store_root / "datasets" / dataset_id).mkdir() + + result = _log_cmd(dataset_id) + + assert result.exit_code == 0, result.output + assert "no snapshots yet" in result.output + + +# --------------------------------------------------------------------------- +# Index building (`_ensure_index`) +# --------------------------------------------------------------------------- + + +def test_log_without_yes_and_without_a_terminal_exits_one( + tmp_path: Path, store_root: Path +) -> None: + """With no index and no tty to ask at, the command says to pass --yes.""" + csv = _write_csv(tmp_path / "data.csv") + _track(store_root, csv) + + result = runner.invoke(app, ["log", str(csv)]) + + assert result.exit_code == 1, result.output + assert "--yes" in result.output + assert not cache.index_exists(store_root) + + +def test_log_prompts_and_builds_when_accepted( + tmp_path: Path, store_root: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """At a terminal the build is offered, and accepting it shows the history.""" + csv = _write_csv(tmp_path / "data.csv") + _track(store_root, csv) + _pretend_terminal(monkeypatch) + + result = runner.invoke(app, ["log", str(csv)], input="y\n") + + assert result.exit_code == 0, result.output + assert "3 rows" in result.output + assert cache.index_exists(store_root) + + +def test_log_prompt_declined_exits_zero_without_building( + tmp_path: Path, store_root: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Declining the build is a choice, not an error, and leaves no index.""" + csv = _write_csv(tmp_path / "data.csv") + _track(store_root, csv) + _pretend_terminal(monkeypatch) + + result = runner.invoke(app, ["log", str(csv)], input="n\n") + + assert result.exit_code == 0, result.output + assert "nothing to show" in result.output + assert not cache.index_exists(store_root) + + +def test_log_existing_index_is_not_offered_again( + tmp_path: Path, store_root: Path +) -> None: + """Once an index exists, no consent is needed to read it.""" + csv = _write_csv(tmp_path / "data.csv") + _track(store_root, csv) + assert _log_cmd(str(csv)).exit_code == 0 + + result = runner.invoke(app, ["log", str(csv)]) + + assert result.exit_code == 0, result.output + assert "no snapshot index" not in result.output.lower() + + +def test_ensure_index_rebuild_discards_the_existing_index( + tmp_path: Path, store_root: Path +) -> None: + """--rebuild deletes the index without asking, implying consent.""" + csv = _write_csv(tmp_path / "data.csv") + _track(store_root, csv) + assert _log_cmd(str(csv)).exit_code == 0 + assert cache.index_exists(store_root) + + _ensure_index(store_root, yes=False, rebuild=True) + + assert not cache.index_exists(store_root) + + +def test_ensure_index_rebuild_on_a_missing_index_is_fine(store_root: Path) -> None: + """--rebuild does not require an index to already be there.""" + _ensure_index(store_root, yes=False, rebuild=True) + + assert not cache.index_exists(store_root) + + +def test_log_rebuild_still_shows_the_history(tmp_path: Path, store_root: Path) -> None: + """The index is rebuilt from the dataset logs, so nothing is lost.""" + csv = _write_csv(tmp_path / "data.csv") + _track(store_root, csv) + assert _log_cmd(str(csv)).exit_code == 0 + + result = runner.invoke(app, ["log", str(csv), "--rebuild"]) + + assert result.exit_code == 0, result.output + assert "3 rows" in result.output + + +def test_ensure_index_without_a_terminal_raises_exit_one(store_root: Path) -> None: + """The no-tty branch exits rather than blocking on a prompt.""" + with pytest.raises(typer.Exit) as excinfo: + _ensure_index(store_root, yes=False, rebuild=False) + + assert excinfo.value.exit_code == 1 + + +# --------------------------------------------------------------------------- +# Target resolution (`_resolve_target`, `_canonical_uuid`) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("label", "spell"), + [ + ("canonical", lambda v: v), + ("uppercased", str.upper), + ("braced", lambda v: f"{{{v}}}"), + ("dashless", lambda v: v.replace("-", "")), + ("urn", lambda v: f"urn:uuid:{v}"), + ], +) +def test_canonical_uuid_accepts_alternate_spellings( + label: str, spell: Callable[[str], str] +) -> None: + """Uppercased, braced, dashless and urn ids all name the same dataset.""" + value = str(uuid.uuid4()) + + assert _canonical_uuid(spell(value)) == value + + +@pytest.mark.parametrize("value", ["", "data.csv", "not-a-uuid", "../etc/passwd"]) +def test_canonical_uuid_rejects_non_uuids(value: str) -> None: + """Anything that is not a UUID in some spelling is not an id.""" + assert _canonical_uuid(value) is None + + +def test_resolve_target_prefers_an_existing_id_over_a_path( + tmp_path: Path, store_root: Path +) -> None: + """An id that names a real dataset wins, so no file can shadow it.""" + csv = _write_csv(tmp_path / "data.csv") + dataset_id = _track(store_root, csv) + cache.sync(store_root) + # A file in the cwd named exactly like the id must not be picked up. + _write_csv(tmp_path / dataset_id) + + resolved = _resolve_target(dataset_id, store_root=store_root, root=None) + + assert resolved == dataset_id + + +def test_resolve_target_falls_back_to_a_path(tmp_path: Path, store_root: Path) -> None: + """A target that is not an existing id is read as a path.""" + csv = _write_csv(tmp_path / "data.csv") + dataset_id = _track(store_root, csv) + cache.sync(store_root) + + assert _resolve_target(str(csv), store_root=store_root, root=None) == dataset_id + + +def test_resolve_target_relative_path_matches(tmp_path: Path, store_root: Path) -> None: + """A relative path resolves the same way an absolute one does.""" + csv = _write_csv(tmp_path / "data.csv") + dataset_id = _track(store_root, csv) + cache.sync(store_root) + + assert _resolve_target("data.csv", store_root=store_root, root=None) == dataset_id + + +def test_resolve_target_unknown_id_reports_the_id(store_root: Path) -> None: + """A well-formed id no dataset goes by is reported as a missing dataset.""" + cache.sync(store_root) + missing = str(uuid.uuid4()) + + with pytest.raises(DatasetNotFoundError, match="No dataset") as excinfo: + _resolve_target(missing, store_root=store_root, root=None) + + assert missing in str(excinfo.value) + + +def test_resolve_target_unknown_path_suggests_the_id(store_root: Path) -> None: + """A path no dataset recorded explains how to recover with a dataset id.""" + cache.sync(store_root) + + with pytest.raises(DatasetNotFoundError, match="No tracked dataset") as excinfo: + _resolve_target("missing.csv", store_root=store_root, root=None) + + message = str(excinfo.value) + assert "missing.csv" in message + + +def test_resolve_target_honours_an_explicit_root( + tmp_path: Path, store_root: Path +) -> None: + """`root` replaces the store root as what the path is made relative to.""" + nested = tmp_path / "nested" + nested.mkdir() + csv = _write_csv(nested / "data.csv") + dataset_id = _track(store_root, csv, "--root", str(nested)) + cache.sync(store_root) + + assert _resolve_target(str(csv), store_root=store_root, root=nested) == dataset_id + with pytest.raises(DatasetNotFoundError): + _resolve_target(str(csv), store_root=store_root, root=None) + + +# --------------------------------------------------------------------------- +# Error listing (`_known_datasets`) +# --------------------------------------------------------------------------- + + +def test_known_datasets_on_an_empty_store(store_root: Path) -> None: + """An empty store says so, and points at the command that fills it.""" + cache.sync(store_root) + + listing = _known_datasets(store_root) + + assert "(none)" in listing + assert "dstrack track" in listing + + +def test_known_datasets_lists_id_name_and_path( + tmp_path: Path, store_root: Path +) -> None: + """Each dataset is shown with what its latest snapshot recorded.""" + csv = _write_csv(tmp_path / "customers.csv") + dataset_id = _track(store_root, csv, "--name", "clients") + cache.sync(store_root) + + listing = _known_datasets(store_root) + + assert dataset_id in listing + assert "clients" in listing + assert "customers.csv" in listing + + +def test_known_datasets_marks_missing_fields(store_root: Path) -> None: + """A dataset with no committed snapshot still appears, with dashes.""" + dataset_id = str(uuid.uuid4()) + (store_root / "datasets" / dataset_id).mkdir() + cache.sync(store_root) + + listing = _known_datasets(store_root) + + assert dataset_id in listing + assert "-" in listing + + +def test_known_datasets_drops_a_dataset_deleted_from_the_store( + tmp_path: Path, store_root: Path +) -> None: + """A dataset removed from the store is forgotten by the next sync. + + The index outlives the logs it was built from, so a dataset directory + deleted by hand would otherwise keep being listed as if it existed. + """ + removed_id = _track(store_root, _write_csv(tmp_path / "gone.csv")) + kept_id = _track(store_root, _write_csv(tmp_path / "kept.csv")) + cache.sync(store_root) + assert removed_id in _known_datasets(store_root) + + shutil.rmtree(store_root / "datasets" / removed_id) + cache.sync(store_root) + + listing = _known_datasets(store_root) + assert removed_id not in listing + assert kept_id in listing + + +def test_log_reports_a_rebuilt_index( + tmp_path: Path, store_root: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A sync that had to rebuild the index says why, and still shows history.""" + csv = _write_csv(tmp_path / "data.csv") + _track(store_root, csv) + assert _log_cmd(str(csv)).exit_code == 0 + monkeypatch.setattr(cache, "sync", lambda root: "the schema changed") + + result = _log_cmd(str(csv)) + + assert result.exit_code == 0, result.output + assert "Rebuilt the snapshot index" in result.output + + +# --------------------------------------------------------------------------- +# Relative timestamps (`_humanize`) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("age", "expected"), + [ + (timedelta(seconds=1), "1 second ago"), + (timedelta(minutes=90), "1 hour ago"), + (timedelta(days=2), "2 days ago"), + (timedelta(days=400), "1 year ago"), + (timedelta(minutes=-1), "in 1 minute"), + (timedelta(days=-14), "in 2 weeks"), + ], +) +def test_humanize_uses_the_largest_unit_that_fits( + age: timedelta, expected: str +) -> None: + """An age reads in whole units, and a future timestamp reads as `in ...`.""" + now = datetime(2026, 7, 20, 12, 0, tzinfo=UTC) + + assert _humanize((now - age).isoformat(), now) == expected + + +def test_log_surfaces_a_corrupt_store( + tmp_path: Path, store_root: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A store-level failure is reported as an error, not a traceback.""" + csv = _write_csv(tmp_path / "data.csv") + dataset_id = _track(store_root, csv) + + def _boom(*args: object, **kwargs: object) -> bool: + raise StoreCorruptionError("log.jsonl is not readable") + + monkeypatch.setattr(_log.store, "dataset_exists", _boom) + + result = _log_cmd(dataset_id) + + assert result.exit_code == 1, result.output + assert "not readable" in result.output diff --git a/tests/test_store.py b/tests/test_store.py index d467826..2b984c4 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -7,8 +7,8 @@ import pytest -from dstrack.errors import DatasetNotFoundError -from dstrack.store import write_snapshot +from dstrack.errors import DatasetNotFoundError, StoreCorruptionError +from dstrack.store import LogEntry, read_log_entries, walk_lineage, write_snapshot def _snapshot(dataset_path: str = "data/ds.csv", **overrides: Any) -> dict[str, Any]: @@ -23,6 +23,26 @@ def _snapshot(dataset_path: str = "data/ds.csv", **overrides: Any) -> dict[str, return snapshot +def _entry(snapshot_id: str, parent: str | None = None) -> LogEntry: + """A log entry carrying only the fields the lineage walk follows.""" + return LogEntry( + snapshot_id=snapshot_id, + parent_snapshot_id=parent, + created_at=None, + created_by=None, + dataset_name=None, + dataset_path=None, + num_rows=None, + num_columns=None, + ) + + +def _rewrite_log(tmp_path: Path, dataset_id: str, *lines: str) -> None: + """Replace a dataset's log with `lines`, each newline-terminated.""" + log_path = tmp_path / "datasets" / dataset_id / "log.jsonl" + log_path.write_text("".join(line + "\n" for line in lines), encoding="utf-8") + + def test_first_snapshot_mints_a_dataset(tmp_path: Path) -> None: """A path no dataset has recorded starts a new lineage with no parent.""" result = write_snapshot(_snapshot(), store_root=tmp_path) @@ -142,3 +162,67 @@ def test_rejected_snapshot_id_writes_nothing(tmp_path: Path) -> None: write_snapshot(_snapshot(snapshot_id="../../evil"), store_root=tmp_path) assert not (tmp_path / "datasets").exists() + + +@pytest.mark.parametrize( + ("bad_line", "message"), + [ + ("{oops", "not valid JSON"), + ('"a string, not an entry"', "not a JSON object"), + ], +) +def test_complete_but_malformed_log_line_is_corruption( + tmp_path: Path, bad_line: str, message: str +) -> None: + """A damaged line that is not the torn final append is reported, not skipped. + + The line is newline-terminated and followed by another, so it was written + whole: an interrupted append could not have produced it. + """ + first = write_snapshot(_snapshot(), store_root=tmp_path) + _rewrite_log(tmp_path, first.dataset_id, bad_line, json.dumps({"snapshot_id": "b"})) + + with pytest.raises(StoreCorruptionError, match=message): + read_log_entries(first.dataset_id, store_root=tmp_path) + + +@pytest.mark.parametrize( + ("line", "message"), + [ + ({"dataset_name": "ds"}, "no usable `snapshot_id`"), + ({"snapshot_id": ""}, "no usable `snapshot_id`"), + ({"snapshot_id": 42}, "no usable `snapshot_id`"), + ({"snapshot_id": "a", "parent_snapshot_id": 7}, "not a snapshot id"), + ], +) +def test_unusable_lineage_fields_are_corruption( + tmp_path: Path, line: dict[str, Any], message: str +) -> None: + """The two fields that drive the lineage walk must be well-typed ids.""" + first = write_snapshot(_snapshot(), store_root=tmp_path) + _rewrite_log(tmp_path, first.dataset_id, json.dumps(line)) + + with pytest.raises(StoreCorruptionError, match=message): + read_log_entries(first.dataset_id, store_root=tmp_path) + + +def test_walk_lineage_missing_head_entry_is_corruption() -> None: + """A HEAD the log has no line for means the log lost the committed entry.""" + with pytest.raises(StoreCorruptionError, match="no entry for snapshot 'head'"): + walk_lineage({}, "head") + + +def test_walk_lineage_missing_ancestor_names_the_child() -> None: + """A broken parent link points at the entry that named the missing one.""" + index = {"child": _entry("child", parent="gone")} + + with pytest.raises(StoreCorruptionError, match="the parent of 'child'"): + walk_lineage(index, "child") + + +def test_walk_lineage_cycle_is_corruption() -> None: + """Parent links edited into a loop terminate the walk instead of hanging.""" + index = {"a": _entry("a", parent="b"), "b": _entry("b", parent="a")} + + with pytest.raises(StoreCorruptionError, match="loops back on itself"): + walk_lineage(index, "a") diff --git a/zensical.toml b/zensical.toml index 618d34d..2073a20 100644 --- a/zensical.toml +++ b/zensical.toml @@ -69,7 +69,7 @@ repo_name = "leoyala/dstrack" # The path provided should be relative to the "docs_dir". # # Read more: https://zensical.org/docs/customization/#additional-javascript -#extra_javascript = ["javascripts/extra.js"] +extra_javascript = ["javascripts/logo-animation.js"] # ---------------------------------------------------------------------------- # Section for configuring theme options