From 4fbc3e6be2103e3216118742444241e5483ca96a Mon Sep 17 00:00:00 2001 From: Will Jones Date: Tue, 30 Jun 2026 13:04:06 -0700 Subject: [PATCH 1/8] test(python): add data overlay benchmark suite Adds a Python benchmark suite for data overlay files in python/ci_benchmarks/, covering three areas: - manifest size growth vs. number of overlays and coverage, - bytes written updating 1% of a column via an overlay vs. update / merge_insert / full rewrite, across narrow and wide rows, - take and scan time + cold read IO as overlay layers, coverage fraction, fragmentation, and data type vary. Shared dataset/overlay construction and measurement helpers live in ci_benchmarks/overlays.py and build fixtures through the public lance.LanceOperation.DataOverlay binding. The write benchmark verifies each approach actually performs the update and splits persisted bytes into data vs. metadata, since the rewrite mechanisms are delete-and- reinsert (whole rows, all columns) while an overlay writes only the changed column. Co-Authored-By: Claude Opus 4.8 (1M context) --- python/python/ci_benchmarks/README.md | 21 ++ .../benchmarks/test_overlay_manifest.py | 51 ++++ .../benchmarks/test_overlay_read.py | 95 +++++++ .../benchmarks/test_overlay_write.py | 163 ++++++++++++ python/python/ci_benchmarks/overlays.py | 231 ++++++++++++++++++ 5 files changed, 561 insertions(+) create mode 100644 python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py create mode 100644 python/python/ci_benchmarks/benchmarks/test_overlay_read.py create mode 100644 python/python/ci_benchmarks/benchmarks/test_overlay_write.py create mode 100644 python/python/ci_benchmarks/overlays.py diff --git a/python/python/ci_benchmarks/README.md b/python/python/ci_benchmarks/README.md index 0245d29166f..ac6f77a6fb7 100644 --- a/python/python/ci_benchmarks/README.md +++ b/python/python/ci_benchmarks/README.md @@ -92,6 +92,27 @@ To save results as JSON (Bencher Metric Format): pytest ... --benchmark-stats-json stats.json ``` +## Data overlay benchmarks + +`benchmarks/test_overlay_*.py` measure the cost of data overlay files (sidecar +files that supply replacement values for a subset of cells, merged on read). +Unlike the other suites they generate their own base datasets and overlays (via +`ci_benchmarks/overlays.py`) into a temporary directory, so no pre-generation step +is needed: + +- `test_overlay_manifest.py` — manifest growth vs. number of overlays and coverage. +- `test_overlay_write.py` — bytes written updating 1% of a column via an overlay vs. + `update` / `merge_insert` / full rewrite. +- `test_overlay_read.py` — take and scan time + cold read IO as overlay layers, + coverage fraction, fragmentation, and data type vary. + +```bash +pytest python/ci_benchmarks/benchmarks/test_overlay_read.py -s +``` + +Size/byte metrics are attached with `record_property` (visible in `--junitxml` output) +and printed with `-s`; read/scan timings use pytest-benchmark (`--benchmark-json`). + ## Investigating memory use for a particular benchmark To investigate memory use for a particular benchmark, you can use the `bytehound` library. diff --git a/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py b/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py new file mode 100644 index 00000000000..b65779fe1d9 --- /dev/null +++ b/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py @@ -0,0 +1,51 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The Lance Authors + +"""Benchmark #1: impact of overlay files on manifest size. + +Each committed overlay adds a ``DataOverlayFile`` entry (a value-file pointer +plus a serialized coverage bitmap) to every overlaid fragment's metadata in the +manifest. This measures how the manifest grows with the number of overlays and +the coverage size, since manifests are read on every dataset open. +""" + +import pytest +from ci_benchmarks.overlays import ( + commit_overlay_layers, + make_base_dataset, + manifest_size, +) + +NUM_ROWS = 1_000_000 +# Single fragment so every overlay lands on the same fragment's metadata. +ROWS_PER_FILE = NUM_ROWS + + +@pytest.mark.parametrize("num_overlays", [0, 1, 4, 16, 64]) +@pytest.mark.parametrize( + "fraction,pattern", + [(0.01, "contiguous"), (0.01, "stride"), (0.1, "stride")], + ids=["1pct-contiguous", "1pct-stride", "10pct-stride"], +) +def test_overlay_manifest_size( + tmp_path, record_property, num_overlays, fraction, pattern +): + base = str(tmp_path / "ds") + ds = make_base_dataset(base, NUM_ROWS, ROWS_PER_FILE, "int32", "2.1") + base_bytes = manifest_size(ds) + + if num_overlays: + ds = commit_overlay_layers(ds, base, num_overlays, fraction, pattern, "int32") + + total = manifest_size(ds) + growth = total - base_bytes + per_overlay = growth / num_overlays if num_overlays else 0 + + record_property("manifest_bytes", total) + record_property("manifest_growth_bytes", growth) + record_property("bytes_per_overlay", per_overlay) + print( + f"\noverlays={num_overlays:>3} {fraction:>5.0%} {pattern:<10} " + f"manifest={total:>8}B growth={growth:>8}B " + f"per_overlay={per_overlay:>8.0f}B" + ) diff --git a/python/python/ci_benchmarks/benchmarks/test_overlay_read.py b/python/python/ci_benchmarks/benchmarks/test_overlay_read.py new file mode 100644 index 00000000000..fdb2f4bfff7 --- /dev/null +++ b/python/python/ci_benchmarks/benchmarks/test_overlay_read.py @@ -0,0 +1,95 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The Lance Authors + +"""Benchmark #3: impact of overlay files on take and scan workloads. + +On read, every overlay covering a requested cell must be consulted and its value +merged over the base. This measures how take and full-scan cost scale with: +- the number of overlay layers stacked on a fragment (compaction payoff), +- coverage fraction (how many cells are overlaid), +- fragmentation (contiguous run vs. strided -> pages touched), and +- data type (int32 vs. a fixed-size-list embedding). + +Wall time is measured warm via pytest-benchmark; read IO (bytes + IOPS) is +measured once cold, after dropping the page cache, via ``io_stats_incremental``. +""" + +import random + +import lance +import pytest +from ci_benchmarks.overlays import commit_overlay_layers, make_base_dataset +from ci_benchmarks.utils import wipe_os_cache + +NUM_ROWS = 1_000_000 +NUM_ROWS_EMBEDDING = 100_000 +ROWS_PER_FILE = NUM_ROWS # single fragment: isolates overlay-layer scaling +TAKE_ROWS = 100 + + +def _take_indices(num_rows: int) -> list: + rng = random.Random(0) + return sorted(rng.sample(range(num_rows), TAKE_ROWS)) + + +def _measure_cold_io(ds: lance.LanceDataset, base: str, work): + """Drop the page cache, run ``work`` once, return its read IO stats.""" + wipe_os_cache(base) + ds.io_stats_incremental() # reset + work() + stats = ds.io_stats_incremental() + return stats.read_bytes, stats.read_iops + + +def _run_read(benchmark, record_property, base, ds, workload, num_rows): + if workload == "take": + indices = _take_indices(num_rows) + + def work(): + ds.take(indices, columns=["val"]) + else: + + def work(): + ds.to_table(columns=["val"]) + + read_bytes, read_iops = _measure_cold_io(ds, base, work) + record_property("cold_read_bytes", read_bytes) + record_property("cold_read_iops", read_iops) + print(f"\ncold read: {read_bytes}B over {read_iops} iops") + + benchmark(work) + + +@pytest.mark.parametrize("version", ["2.0", "2.1"]) +@pytest.mark.parametrize("workload", ["take", "scan"]) +@pytest.mark.parametrize("num_overlays", [0, 4, 16]) +@pytest.mark.parametrize( + "fraction,pattern", + [(0.01, "contiguous"), (0.01, "stride")], + ids=["1pct-contiguous", "1pct-stride"], +) +def test_overlay_read_scaling( + benchmark, + tmp_path, + record_property, + version, + workload, + num_overlays, + fraction, + pattern, +): + base = str(tmp_path / "ds") + ds = make_base_dataset(base, NUM_ROWS, ROWS_PER_FILE, "int32", version) + if num_overlays: + ds = commit_overlay_layers(ds, base, num_overlays, fraction, pattern, "int32") + _run_read(benchmark, record_property, base, ds, workload, NUM_ROWS) + + +@pytest.mark.parametrize("workload", ["take", "scan"]) +@pytest.mark.parametrize("dtype", ["int32", "embedding"]) +def test_overlay_read_dtype(benchmark, tmp_path, record_property, workload, dtype): + num_rows = NUM_ROWS_EMBEDDING if dtype == "embedding" else NUM_ROWS + base = str(tmp_path / "ds") + ds = make_base_dataset(base, num_rows, num_rows, dtype, "2.1") + ds = commit_overlay_layers(ds, base, 4, 0.01, "stride", dtype) + _run_read(benchmark, record_property, base, ds, workload, num_rows) diff --git a/python/python/ci_benchmarks/benchmarks/test_overlay_write.py b/python/python/ci_benchmarks/benchmarks/test_overlay_write.py new file mode 100644 index 00000000000..bf8dd0e191f --- /dev/null +++ b/python/python/ci_benchmarks/benchmarks/test_overlay_write.py @@ -0,0 +1,163 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The Lance Authors + +"""Benchmark #2: cost of updating a single column for 1% of rows. + +Compares writing an overlay against the existing rewrite-based mechanisms +(``update``, ``merge_insert``, full overwrite). + +The rewrite mechanisms are delete-and-reinsert: they re-encode the changed rows +*across all columns* into a new fragment and add a deletion vector to each +affected fragment. An overlay instead persists only the changed column's cells +plus a coverage bitmap in the manifest. The crossover therefore depends on row +width, so the benchmark sweeps a ``width`` axis: + +- ``narrow``: id + val only. Rewriting whole rows is nearly free, so an overlay + has no byte advantage and its manifest/bitmap overhead can make it larger. +- ``wide``: id + val + a 64-d float32 payload. Rewrites re-encode the payload for + every changed row; the overlay still writes only ``val``. + +Persisted bytes are split into data (under ``data/``) and metadata (manifests, +transactions, deletion vectors), because the two families spend their bytes +differently. ``read_bytes`` (process-wide read syscalls from /proc/self/io) and +wall time are reported alongside: the rewrite mechanisms must read the affected +rows to re-encode them, which both costs IO and explains much of their extra +time, while an overlay reads essentially nothing. + +NOTE: the overlay arm is hand-rolled (no ``update``-via-overlay path exists yet) +and, unlike the other arms, does not read the base column first -- it writes +fresh values for the covered cells. That no-read is the win this read metric +exposes. +""" + +from time import perf_counter + +import lance +import numpy as np +import pyarrow as pa +import pytest +from ci_benchmarks.overlays import ( + commit_overlay_layers, + file_sizes, + make_base_dataset, + proc_io_counters, + written_breakdown, +) + +NUM_ROWS = 1_000_000 +ROWS_PER_FILE = 100_000 # 10 fragments; the strided selection hits every one +UPDATE_FRACTION = 0.01 +STEP = int(1 / UPDATE_FRACTION) # select id % STEP == 0 +WIDE_DIM = 64 + +APPROACHES = [ + "update", + "merge_insert", + "merge_insert_indexed", + "overlay", + "full_rewrite", +] + + +def _selected_ids() -> np.ndarray: + return np.arange(0, NUM_ROWS, STEP, dtype=np.int64) + + +def _payload(n: int, rng: np.random.Generator) -> pa.Array: + flat = rng.random(n * WIDE_DIM, dtype=np.float32) + return pa.FixedSizeListArray.from_arrays(pa.array(flat), WIDE_DIM) + + +def _new_rows(ids: np.ndarray, payload_dim: int) -> pa.Table: + """Full replacement rows for ``merge_insert`` (a row-oriented op cannot + target a single column, so it brings whole rows).""" + rng = np.random.default_rng(1) + cols = { + "id": pa.array(ids), + "val": pa.array(rng.integers(0, 1 << 30, len(ids), np.int32)), + } + if payload_dim: + cols["payload"] = _payload(len(ids), rng) + return pa.table(cols) + + +@pytest.mark.parametrize("width", ["narrow", "wide"]) +@pytest.mark.parametrize("approach", APPROACHES) +def test_overlay_write_cost(tmp_path, record_property, approach, width): + payload_dim = WIDE_DIM if width == "wide" else 0 + base = str(tmp_path / "ds") + ds = make_base_dataset( + base, NUM_ROWS, ROWS_PER_FILE, "int32", "2.1", payload_dim=payload_dim + ) + ids = _selected_ids() + sel, unsel = int(ids[0]), int(ids[0]) + 1 + before = { + r["id"]: r["val"] + for r in ds.to_table( + columns=["id", "val"], filter=f"id IN ({sel},{unsel})" + ).to_pylist() + } + + if approach == "merge_insert_indexed": + ds.create_scalar_index("id", "BTREE") # setup, not measured + + def run(): + nonlocal ds + if approach == "update": + ds.update({"val": "val + 1"}, where=f"id % {STEP} = 0") + elif approach in ("merge_insert", "merge_insert_indexed"): + ds.merge_insert("id").when_matched_update_all().execute( + _new_rows(ids, payload_dim) + ) + elif approach == "overlay": + ds = commit_overlay_layers(ds, base, 1, UPDATE_FRACTION, "stride", "int32") + elif approach == "full_rewrite": + table = ds.to_table() + val = table.column("val").to_numpy(zero_copy_only=False).copy() + val[ids] = _new_rows(ids, 0).column("val").to_numpy() + cols = {"id": table.column("id"), "val": pa.array(val)} + if payload_dim: + cols["payload"] = table.column("payload") + ds = lance.write_dataset( + pa.table(cols), base, mode="overwrite", data_storage_version="2.1" + ) + else: + raise ValueError(approach) + + snap_before = file_sizes(base) + io_before = proc_io_counters() + start = perf_counter() + run() + elapsed = perf_counter() - start + io_after = proc_io_counters() + snap_after = file_sizes(base) + + # Guard against silently measuring a no-op: the selected row must change and + # the unselected row must not. + ds = lance.dataset(base) + after = { + r["id"]: r["val"] + for r in ds.to_table( + columns=["id", "val"], filter=f"id IN ({sel},{unsel})" + ).to_pylist() + } + assert after[sel] != before[sel], f"{approach}/{width}: selected row unchanged" + assert after[unsel] == before[unsel], f"{approach}/{width}: unselected row changed" + + data_bytes, meta_bytes = written_breakdown(snap_before, snap_after) + # rchar/wchar (syscall bytes) are cache-independent; the base is warm from + # the build, so physical read_bytes would under-report what was read. + read_bytes = ( + io_after["rchar"] - io_before["rchar"] if io_before and io_after else -1 + ) + + record_property("data_bytes", data_bytes) + record_property("metadata_bytes", meta_bytes) + record_property("persisted_bytes", data_bytes + meta_bytes) + record_property("read_bytes", read_bytes) + record_property("seconds", elapsed) + print( + f"\n{approach:<22} {width:<7} read={read_bytes:>11}B " + f"data={data_bytes:>10}B meta={meta_bytes:>9}B " + f"persisted={data_bytes + meta_bytes:>10}B time={elapsed:>7.3f}s" + ) diff --git a/python/python/ci_benchmarks/overlays.py b/python/python/ci_benchmarks/overlays.py new file mode 100644 index 00000000000..0ebb49e9b6c --- /dev/null +++ b/python/python/ci_benchmarks/overlays.py @@ -0,0 +1,231 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The Lance Authors + +"""Shared helpers for the data-overlay benchmark suite. + +Data overlay files supply replacement values for a subset of (row offset, field) +cells in a fragment, merged on read, without rewriting the base data file. These +helpers build synthetic base datasets, commit overlay layers through the public +``lance.LanceOperation.DataOverlay`` operation, and measure their cost. +""" + +import os + +# Data overlay support is gated off in release builds unless this is set (it is +# always on in debug builds). Benchmarks usually run against a release build, so +# enable it here, before lance is imported, or reading overlay datasets fails. +os.environ.setdefault("LANCE_ENABLE_DATA_OVERLAY_FILES", "1") + +from typing import List, Optional, Tuple # noqa: E402 +from urllib.parse import urlparse # noqa: E402 + +import lance # noqa: E402 +import numpy as np # noqa: E402 +import pyarrow as pa # noqa: E402 +from lance.file import LanceFileWriter # noqa: E402 + +EMBEDDING_DIM = 128 + + +def value_type(dtype: str) -> pa.DataType: + if dtype == "int32": + return pa.int32() + if dtype == "embedding": + return pa.list_(pa.float32(), EMBEDDING_DIM) + raise ValueError(f"unknown overlay benchmark dtype {dtype!r}") + + +def _gen_values(dtype: str, n: int, rng: np.random.Generator) -> pa.Array: + if dtype == "int32": + return pa.array(rng.integers(0, 1 << 30, size=n, dtype=np.int32)) + if dtype == "embedding": + flat = rng.random(n * EMBEDDING_DIM, dtype=np.float32) + return pa.FixedSizeListArray.from_arrays(pa.array(flat), EMBEDDING_DIM) + raise ValueError(f"unknown overlay benchmark dtype {dtype!r}") + + +def make_base_dataset( + base_path: str, + num_rows: int, + rows_per_file: int, + dtype: str, + version: str, + payload_dim: int = 0, +) -> lance.LanceDataset: + """Create a base dataset with an ``id`` key and a ``val`` payload column. + + ``val`` is the column overlays target. ``rows_per_file`` controls the number + of fragments (``num_rows // rows_per_file``). When ``payload_dim`` > 0 an + extra ``payload`` fixed-size-list float32 column of that width is added, + making rows wider so the cost of rewriting whole rows (update / merge_insert) + can be compared against overlaying just ``val``. + """ + vtype = value_type(dtype) + fields = {"id": pa.int64(), "val": vtype} + if payload_dim: + fields["payload"] = pa.list_(pa.float32(), payload_dim) + schema = pa.schema(fields) + rng = np.random.default_rng(0) + + def batches(): + written = 0 + while written < num_rows: + n = min(rows_per_file, num_rows - written) + ids = pa.array(range(written, written + n), pa.int64()) + cols = [ids, _gen_values(dtype, n, rng)] + if payload_dim: + flat = rng.random(n * payload_dim, dtype=np.float32) + cols.append( + pa.FixedSizeListArray.from_arrays(pa.array(flat), payload_dim) + ) + yield pa.record_batch(cols, schema=schema) + written += n + + reader = pa.RecordBatchReader.from_batches(schema, batches()) + return lance.write_dataset( + reader, + base_path, + schema=schema, + max_rows_per_file=rows_per_file, + data_storage_version=version, + ) + + +def coverage_offsets(num_rows: int, fraction: float, pattern: str) -> List[int]: + """Offsets within a fragment covered by an overlay. + + ``contiguous`` packs the covered cells into a single leading run (few pages + touched); ``stride`` spreads them evenly across the fragment (many pages + touched). Both cover ``round(num_rows * fraction)`` cells. + """ + count = max(1, int(round(num_rows * fraction))) + if pattern == "contiguous": + return list(range(count)) + if pattern == "stride": + step = max(1, num_rows // count) + return list(range(0, num_rows, step))[:count] + raise ValueError(f"unknown coverage pattern {pattern!r}") + + +def _val_field_id(ds: lance.LanceDataset) -> int: + base_df = ds.get_fragments()[0].metadata.files[0] + names = [f.name for f in ds.schema] + return base_df.fields[names.index("val")] + + +def commit_overlay_layers( + ds: lance.LanceDataset, + base_path: str, + num_layers: int, + fraction: float, + pattern: str, + dtype: str, + *, + seed: int = 0, +) -> lance.LanceDataset: + """Commit ``num_layers`` overlays on ``val``, each covering the same offsets + in every fragment so that all layers must be consulted on read (the case + that motivates compaction). Returns the updated dataset. + """ + base_df = ds.get_fragments()[0].metadata.files[0] + field_id = _val_field_id(ds) + data_dir = os.path.join(_local_path(ds), "data") + for layer in range(num_layers): + rng = np.random.default_rng(seed + layer + 1) + groups = [] + for frag in ds.get_fragments(): + offsets = coverage_offsets(frag.count_rows(), fraction, pattern) + values = _gen_values(dtype, len(offsets), rng) + batch = pa.record_batch([values], names=["val"]) + name = f"overlay_l{layer}_f{frag.fragment_id}.lance" + path = os.path.join(data_dir, name) + with LanceFileWriter(path) as writer: + writer.write_batch(batch) + df = lance.fragment.DataFile( + path=name, + fields=[field_id], + column_indices=[0], + file_major_version=base_df.file_major_version, + file_minor_version=base_df.file_minor_version, + file_size_bytes=os.path.getsize(path), + ) + groups.append( + lance.LanceOperation.DataOverlayGroup( + fragment_id=frag.fragment_id, + overlays=[ + lance.LanceOperation.DataOverlayFile( + data_file=df, shared_offsets=offsets + ) + ], + ) + ) + op = lance.LanceOperation.DataOverlay(groups=groups) + ds = lance.LanceDataset.commit(ds, op, read_version=ds.version) + return ds + + +# --- Measurement helpers ---------------------------------------------------- + + +def _local_path(ds: lance.LanceDataset) -> str: + parsed = urlparse(ds.uri) + return parsed.path if parsed.scheme == "file" else ds.uri + + +def proc_io_counters() -> Optional[dict]: + """Process-wide /proc/self/io counters, or None if unavailable. + + Captures IO regardless of which object store issued it, which is what makes + it the right tool for comparing mechanisms whose IO does not all flow through + a single dataset handle. Use ``rchar``/``wchar`` (bytes moved by read/write + syscalls) rather than ``read_bytes``/``write_bytes`` (physical storage IO): + freshly written data sits in the page cache, so physical reads under-report + what an operation actually read. + """ + try: + stats = {} + with open("/proc/self/io") as f: + for line in f: + key, _, value = line.partition(":") + stats[key.strip()] = int(value) + return stats + except OSError: + return None + + +def file_sizes(path: str) -> dict: + """Map of relative-path -> size in bytes for every file under ``path``.""" + out = {} + for root, _, files in os.walk(path): + for name in files: + fp = os.path.join(root, name) + if os.path.isfile(fp): + out[os.path.relpath(fp, path)] = os.path.getsize(fp) + return out + + +def dir_size(path: str) -> int: + """Total bytes of all files under ``path`` (persisted storage cost).""" + return sum(file_sizes(path).values()) + + +def written_breakdown(before: dict, after: dict) -> Tuple[int, int]: + """Split newly written/grown bytes (``after`` vs ``before`` snapshots) into + (data_bytes, metadata_bytes). Data is everything under ``data/``; metadata is + manifests, transactions, and deletion files.""" + data = meta = 0 + for path, size in after.items(): + delta = size - before.get(path, 0) + if delta <= 0: + continue + if path.startswith("data" + os.sep): + data += delta + else: + meta += delta + return data, meta + + +def manifest_size(ds: lance.LanceDataset) -> int: + """Size in bytes of the manifest for the dataset's current version.""" + name = f"{(1 << 64) - 1 - ds.version}.manifest" + return os.path.getsize(os.path.join(_local_path(ds), "_versions", name)) From 60dab05bd3f140ba3cf229ba7574570446d8b507 Mon Sep 17 00:00:00 2001 From: Will Jones Date: Thu, 2 Jul 2026 12:49:02 -0700 Subject: [PATCH 2/8] test(python): add wide-column (3072-d embedding) overlay read benchmark The overlay read benchmark's take/scan scaling sweep only covered a 4-byte int32 value column, where the base read is trivial and per-layer merge cost dominates. ML overlays typically target wide vector columns, so add test_overlay_read_wide: the same overlays x coverage-pattern grid on a 3072-d float32 embedding (12 KiB/row). Thread an embedding_dim through the overlay helpers so the value width is configurable. Co-Authored-By: Claude Opus 4.8 (1M context) --- python/python/ci_benchmarks/README.md | 3 +- .../benchmarks/test_overlay_read.py | 49 +++++++++++++++++++ python/python/ci_benchmarks/overlays.py | 35 ++++++++----- 3 files changed, 74 insertions(+), 13 deletions(-) diff --git a/python/python/ci_benchmarks/README.md b/python/python/ci_benchmarks/README.md index ac6f77a6fb7..9ecd1090521 100644 --- a/python/python/ci_benchmarks/README.md +++ b/python/python/ci_benchmarks/README.md @@ -104,7 +104,8 @@ is needed: - `test_overlay_write.py` — bytes written updating 1% of a column via an overlay vs. `update` / `merge_insert` / full rewrite. - `test_overlay_read.py` — take and scan time + cold read IO as overlay layers, - coverage fraction, fragmentation, and data type vary. + coverage fraction, fragmentation, and value width (narrow `int32` vs. a wide + 3072-d embedding) vary. ```bash pytest python/ci_benchmarks/benchmarks/test_overlay_read.py -s diff --git a/python/python/ci_benchmarks/benchmarks/test_overlay_read.py b/python/python/ci_benchmarks/benchmarks/test_overlay_read.py index fdb2f4bfff7..67c30dc8bdb 100644 --- a/python/python/ci_benchmarks/benchmarks/test_overlay_read.py +++ b/python/python/ci_benchmarks/benchmarks/test_overlay_read.py @@ -26,6 +26,13 @@ ROWS_PER_FILE = NUM_ROWS # single fragment: isolates overlay-layer scaling TAKE_ROWS = 100 +# Wide value column: a 3072-d float32 embedding is 12 KiB/row, ~750x an int32 +# cell. Fewer rows keep the base file to ~1.2 GiB while each cell still dominates +# read cost, so the merge/interleave a scan pays per overlay layer moves real +# payload rather than 4-byte integers. +WIDE_EMBEDDING_DIM = 3072 +NUM_ROWS_WIDE = 100_000 + def _take_indices(num_rows: int) -> list: rng = random.Random(0) @@ -93,3 +100,45 @@ def test_overlay_read_dtype(benchmark, tmp_path, record_property, workload, dtyp ds = make_base_dataset(base, num_rows, num_rows, dtype, "2.1") ds = commit_overlay_layers(ds, base, 4, 0.01, "stride", dtype) _run_read(benchmark, record_property, base, ds, workload, num_rows) + + +# Mirror test_overlay_read_scaling but on a wide 3072-d embedding column, so the +# take/scan-vs-layers story can be read for a fat value column rather than a +# 4-byte one. Pinned to the 2.1 format (v2.0 layer scaling is already covered by +# the narrow scan above; the wide-column question is about per-layer payload). +@pytest.mark.parametrize("workload", ["take", "scan"]) +@pytest.mark.parametrize("num_overlays", [0, 4, 16]) +@pytest.mark.parametrize( + "fraction,pattern", + [(0.01, "contiguous"), (0.01, "stride")], + ids=["1pct-contiguous", "1pct-stride"], +) +def test_overlay_read_wide( + benchmark, + tmp_path, + record_property, + workload, + num_overlays, + fraction, + pattern, +): + base = str(tmp_path / "ds") + ds = make_base_dataset( + base, + NUM_ROWS_WIDE, + NUM_ROWS_WIDE, + "embedding", + "2.1", + embedding_dim=WIDE_EMBEDDING_DIM, + ) + if num_overlays: + ds = commit_overlay_layers( + ds, + base, + num_overlays, + fraction, + pattern, + "embedding", + embedding_dim=WIDE_EMBEDDING_DIM, + ) + _run_read(benchmark, record_property, base, ds, workload, NUM_ROWS_WIDE) diff --git a/python/python/ci_benchmarks/overlays.py b/python/python/ci_benchmarks/overlays.py index 0ebb49e9b6c..77687bc7828 100644 --- a/python/python/ci_benchmarks/overlays.py +++ b/python/python/ci_benchmarks/overlays.py @@ -24,23 +24,30 @@ import pyarrow as pa # noqa: E402 from lance.file import LanceFileWriter # noqa: E402 +# Default width for the ``embedding`` dtype. Individual benchmarks override it +# via ``embedding_dim`` to model narrow vs. wide (e.g. 3072-d) value columns. EMBEDDING_DIM = 128 -def value_type(dtype: str) -> pa.DataType: +def value_type(dtype: str, embedding_dim: int = EMBEDDING_DIM) -> pa.DataType: if dtype == "int32": return pa.int32() if dtype == "embedding": - return pa.list_(pa.float32(), EMBEDDING_DIM) + return pa.list_(pa.float32(), embedding_dim) raise ValueError(f"unknown overlay benchmark dtype {dtype!r}") -def _gen_values(dtype: str, n: int, rng: np.random.Generator) -> pa.Array: +def _gen_values( + dtype: str, + n: int, + rng: np.random.Generator, + embedding_dim: int = EMBEDDING_DIM, +) -> pa.Array: if dtype == "int32": return pa.array(rng.integers(0, 1 << 30, size=n, dtype=np.int32)) if dtype == "embedding": - flat = rng.random(n * EMBEDDING_DIM, dtype=np.float32) - return pa.FixedSizeListArray.from_arrays(pa.array(flat), EMBEDDING_DIM) + flat = rng.random(n * embedding_dim, dtype=np.float32) + return pa.FixedSizeListArray.from_arrays(pa.array(flat), embedding_dim) raise ValueError(f"unknown overlay benchmark dtype {dtype!r}") @@ -51,16 +58,19 @@ def make_base_dataset( dtype: str, version: str, payload_dim: int = 0, + embedding_dim: int = EMBEDDING_DIM, ) -> lance.LanceDataset: """Create a base dataset with an ``id`` key and a ``val`` payload column. ``val`` is the column overlays target. ``rows_per_file`` controls the number - of fragments (``num_rows // rows_per_file``). When ``payload_dim`` > 0 an - extra ``payload`` fixed-size-list float32 column of that width is added, - making rows wider so the cost of rewriting whole rows (update / merge_insert) - can be compared against overlaying just ``val``. + of fragments (``num_rows // rows_per_file``). ``embedding_dim`` sets the width + of the ``val`` column when ``dtype`` is ``embedding`` (e.g. 3072 for a wide + embedding). When ``payload_dim`` > 0 an extra ``payload`` fixed-size-list + float32 column of that width is added, making rows wider so the cost of + rewriting whole rows (update / merge_insert) can be compared against + overlaying just ``val``. """ - vtype = value_type(dtype) + vtype = value_type(dtype, embedding_dim) fields = {"id": pa.int64(), "val": vtype} if payload_dim: fields["payload"] = pa.list_(pa.float32(), payload_dim) @@ -72,7 +82,7 @@ def batches(): while written < num_rows: n = min(rows_per_file, num_rows - written) ids = pa.array(range(written, written + n), pa.int64()) - cols = [ids, _gen_values(dtype, n, rng)] + cols = [ids, _gen_values(dtype, n, rng, embedding_dim)] if payload_dim: flat = rng.random(n * payload_dim, dtype=np.float32) cols.append( @@ -122,6 +132,7 @@ def commit_overlay_layers( dtype: str, *, seed: int = 0, + embedding_dim: int = EMBEDDING_DIM, ) -> lance.LanceDataset: """Commit ``num_layers`` overlays on ``val``, each covering the same offsets in every fragment so that all layers must be consulted on read (the case @@ -135,7 +146,7 @@ def commit_overlay_layers( groups = [] for frag in ds.get_fragments(): offsets = coverage_offsets(frag.count_rows(), fraction, pattern) - values = _gen_values(dtype, len(offsets), rng) + values = _gen_values(dtype, len(offsets), rng, embedding_dim) batch = pa.record_batch([values], names=["val"]) name = f"overlay_l{layer}_f{frag.fragment_id}.lance" path = os.path.join(data_dir, name) From 388bf194e8abefb37882102ec78e0fc49f864ca3 Mon Sep 17 00:00:00 2001 From: Will Jones Date: Mon, 20 Jul 2026 17:24:30 -0700 Subject: [PATCH 3/8] fix(python): match overlay benchmark helper to the actual DataOverlay API `DataOverlayFile` takes a single `offsets` param, not `shared_offsets`, and the release-build opt-in env var is `LANCE_ENABLE_UNSTABLE_DATA_OVERLAY_FILES` (not `LANCE_ENABLE_DATA_OVERLAY_FILES`). Both diverged from the benchmark helper after PR #7540 landed the real Python bindings; commit()'s returned handle skipped flag validation so this only surfaced on a fresh dataset open. --- python/python/ci_benchmarks/overlays.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/python/python/ci_benchmarks/overlays.py b/python/python/ci_benchmarks/overlays.py index 77687bc7828..8a6df9e6f9c 100644 --- a/python/python/ci_benchmarks/overlays.py +++ b/python/python/ci_benchmarks/overlays.py @@ -14,7 +14,7 @@ # Data overlay support is gated off in release builds unless this is set (it is # always on in debug builds). Benchmarks usually run against a release build, so # enable it here, before lance is imported, or reading overlay datasets fails. -os.environ.setdefault("LANCE_ENABLE_DATA_OVERLAY_FILES", "1") +os.environ.setdefault("LANCE_ENABLE_UNSTABLE_DATA_OVERLAY_FILES", "1") from typing import List, Optional, Tuple # noqa: E402 from urllib.parse import urlparse # noqa: E402 @@ -165,7 +165,7 @@ def commit_overlay_layers( fragment_id=frag.fragment_id, overlays=[ lance.LanceOperation.DataOverlayFile( - data_file=df, shared_offsets=offsets + data_file=df, offsets=offsets ) ], ) From 8204329cc101e1ca1c11c97d313deb297ab49e0b Mon Sep 17 00:00:00 2001 From: Will Jones Date: Thu, 23 Jul 2026 09:53:18 -0700 Subject: [PATCH 4/8] =?UTF-8?q?test(python):=20polish=20overlay=20benchmar?= =?UTF-8?q?ks=20=E2=80=94=20drop=20dead=20code,=20guard=20fixtures?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review pass on the overlay benchmark suite: - Remove the unused `base_path` parameter of `commit_overlay_layers` (the body derives its data dir from the dataset URI) and the dead `dir_size` helper; both had no effect and misled the reader. - Rename `value_type` -> `_value_type` to match the module's private helpers. - Document the `{u64::MAX - version}.manifest` naming in `manifest_size`. - Guard the read and manifest benchmark fixtures: assert a covered cell reads back the overlaid value and that overlays grow the manifest, so a silently-broken overlay can't let the benchmark pass while measuring plain base reads. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../benchmarks/test_overlay_manifest.py | 6 +++++- .../benchmarks/test_overlay_read.py | 21 ++++++++++++++++--- .../benchmarks/test_overlay_write.py | 2 +- python/python/ci_benchmarks/overlays.py | 12 ++++------- 4 files changed, 28 insertions(+), 13 deletions(-) diff --git a/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py b/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py index b65779fe1d9..8339635c8df 100644 --- a/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py +++ b/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py @@ -35,10 +35,14 @@ def test_overlay_manifest_size( base_bytes = manifest_size(ds) if num_overlays: - ds = commit_overlay_layers(ds, base, num_overlays, fraction, pattern, "int32") + ds = commit_overlay_layers(ds, num_overlays, fraction, pattern, "int32") total = manifest_size(ds) growth = total - base_bytes + # Guard the fixture: committed overlays must enlarge the manifest, else the + # benchmark would report growth=0 for overlays that were never recorded. + if num_overlays: + assert growth > 0, "overlays did not grow the manifest" per_overlay = growth / num_overlays if num_overlays else 0 record_property("manifest_bytes", total) diff --git a/python/python/ci_benchmarks/benchmarks/test_overlay_read.py b/python/python/ci_benchmarks/benchmarks/test_overlay_read.py index 67c30dc8bdb..edafc6813f4 100644 --- a/python/python/ci_benchmarks/benchmarks/test_overlay_read.py +++ b/python/python/ci_benchmarks/benchmarks/test_overlay_read.py @@ -39,6 +39,16 @@ def _take_indices(num_rows: int) -> list: return sorted(rng.sample(range(num_rows), TAKE_ROWS)) +def _covered_value(ds: lance.LanceDataset): + """``val`` at offset 0, which every coverage pattern includes. + + Used to guard the fixture: after committing overlays a covered cell must + read back a new value, otherwise a read-invisible overlay would let the + benchmark silently time plain base reads. + """ + return ds.take([0], columns=["val"]).column("val").to_pylist()[0] + + def _measure_cold_io(ds: lance.LanceDataset, base: str, work): """Drop the page cache, run ``work`` once, return its read IO stats.""" wipe_os_cache(base) @@ -88,7 +98,9 @@ def test_overlay_read_scaling( base = str(tmp_path / "ds") ds = make_base_dataset(base, NUM_ROWS, ROWS_PER_FILE, "int32", version) if num_overlays: - ds = commit_overlay_layers(ds, base, num_overlays, fraction, pattern, "int32") + base_val = _covered_value(ds) + ds = commit_overlay_layers(ds, num_overlays, fraction, pattern, "int32") + assert _covered_value(ds) != base_val, "overlay not visible on read" _run_read(benchmark, record_property, base, ds, workload, NUM_ROWS) @@ -98,7 +110,9 @@ def test_overlay_read_dtype(benchmark, tmp_path, record_property, workload, dtyp num_rows = NUM_ROWS_EMBEDDING if dtype == "embedding" else NUM_ROWS base = str(tmp_path / "ds") ds = make_base_dataset(base, num_rows, num_rows, dtype, "2.1") - ds = commit_overlay_layers(ds, base, 4, 0.01, "stride", dtype) + base_val = _covered_value(ds) + ds = commit_overlay_layers(ds, 4, 0.01, "stride", dtype) + assert _covered_value(ds) != base_val, "overlay not visible on read" _run_read(benchmark, record_property, base, ds, workload, num_rows) @@ -132,13 +146,14 @@ def test_overlay_read_wide( embedding_dim=WIDE_EMBEDDING_DIM, ) if num_overlays: + base_val = _covered_value(ds) ds = commit_overlay_layers( ds, - base, num_overlays, fraction, pattern, "embedding", embedding_dim=WIDE_EMBEDDING_DIM, ) + assert _covered_value(ds) != base_val, "overlay not visible on read" _run_read(benchmark, record_property, base, ds, workload, NUM_ROWS_WIDE) diff --git a/python/python/ci_benchmarks/benchmarks/test_overlay_write.py b/python/python/ci_benchmarks/benchmarks/test_overlay_write.py index bf8dd0e191f..0a54c8a3ac7 100644 --- a/python/python/ci_benchmarks/benchmarks/test_overlay_write.py +++ b/python/python/ci_benchmarks/benchmarks/test_overlay_write.py @@ -110,7 +110,7 @@ def run(): _new_rows(ids, payload_dim) ) elif approach == "overlay": - ds = commit_overlay_layers(ds, base, 1, UPDATE_FRACTION, "stride", "int32") + ds = commit_overlay_layers(ds, 1, UPDATE_FRACTION, "stride", "int32") elif approach == "full_rewrite": table = ds.to_table() val = table.column("val").to_numpy(zero_copy_only=False).copy() diff --git a/python/python/ci_benchmarks/overlays.py b/python/python/ci_benchmarks/overlays.py index 8a6df9e6f9c..523a17ed613 100644 --- a/python/python/ci_benchmarks/overlays.py +++ b/python/python/ci_benchmarks/overlays.py @@ -29,7 +29,7 @@ EMBEDDING_DIM = 128 -def value_type(dtype: str, embedding_dim: int = EMBEDDING_DIM) -> pa.DataType: +def _value_type(dtype: str, embedding_dim: int = EMBEDDING_DIM) -> pa.DataType: if dtype == "int32": return pa.int32() if dtype == "embedding": @@ -70,7 +70,7 @@ def make_base_dataset( rewriting whole rows (update / merge_insert) can be compared against overlaying just ``val``. """ - vtype = value_type(dtype, embedding_dim) + vtype = _value_type(dtype, embedding_dim) fields = {"id": pa.int64(), "val": vtype} if payload_dim: fields["payload"] = pa.list_(pa.float32(), payload_dim) @@ -125,7 +125,6 @@ def _val_field_id(ds: lance.LanceDataset) -> int: def commit_overlay_layers( ds: lance.LanceDataset, - base_path: str, num_layers: int, fraction: float, pattern: str, @@ -215,11 +214,6 @@ def file_sizes(path: str) -> dict: return out -def dir_size(path: str) -> int: - """Total bytes of all files under ``path`` (persisted storage cost).""" - return sum(file_sizes(path).values()) - - def written_breakdown(before: dict, after: dict) -> Tuple[int, int]: """Split newly written/grown bytes (``after`` vs ``before`` snapshots) into (data_bytes, metadata_bytes). Data is everything under ``data/``; metadata is @@ -238,5 +232,7 @@ def written_breakdown(before: dict, after: dict) -> Tuple[int, int]: def manifest_size(ds: lance.LanceDataset) -> int: """Size in bytes of the manifest for the dataset's current version.""" + # Manifests are named `{u64::MAX - version}.manifest` so that a plain + # lexicographic directory listing yields newest-version-first. name = f"{(1 << 64) - 1 - ds.version}.manifest" return os.path.getsize(os.path.join(_local_path(ds), "_versions", name)) From fe36840d028e916a6e9408be412341447e6526e9 Mon Sep 17 00:00:00 2001 From: Will Jones Date: Thu, 23 Jul 2026 10:18:40 -0700 Subject: [PATCH 5/8] =?UTF-8?q?test(python):=20address=20review=20?= =?UTF-8?q?=E2=80=94=20trim=20README,=20drop=20prints,=20cut=20redundant?= =?UTF-8?q?=20read=20benchmark?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - README: remove the per-benchmark descriptions; keep it to general suite advice. - Manifest/read benchmarks: report via record_property only, no printing. - Drop test_overlay_read_dtype: its int32 and embedding read costs are already covered by test_overlay_read_scaling (int32) and test_overlay_read_wide (embedding) across the full overlay-layer sweep. Co-Authored-By: Claude Opus 4.8 (1M context) --- python/python/ci_benchmarks/README.md | 22 ------------------- .../benchmarks/test_overlay_manifest.py | 5 ----- .../benchmarks/test_overlay_read.py | 16 +------------- 3 files changed, 1 insertion(+), 42 deletions(-) diff --git a/python/python/ci_benchmarks/README.md b/python/python/ci_benchmarks/README.md index 9ecd1090521..0245d29166f 100644 --- a/python/python/ci_benchmarks/README.md +++ b/python/python/ci_benchmarks/README.md @@ -92,28 +92,6 @@ To save results as JSON (Bencher Metric Format): pytest ... --benchmark-stats-json stats.json ``` -## Data overlay benchmarks - -`benchmarks/test_overlay_*.py` measure the cost of data overlay files (sidecar -files that supply replacement values for a subset of cells, merged on read). -Unlike the other suites they generate their own base datasets and overlays (via -`ci_benchmarks/overlays.py`) into a temporary directory, so no pre-generation step -is needed: - -- `test_overlay_manifest.py` — manifest growth vs. number of overlays and coverage. -- `test_overlay_write.py` — bytes written updating 1% of a column via an overlay vs. - `update` / `merge_insert` / full rewrite. -- `test_overlay_read.py` — take and scan time + cold read IO as overlay layers, - coverage fraction, fragmentation, and value width (narrow `int32` vs. a wide - 3072-d embedding) vary. - -```bash -pytest python/ci_benchmarks/benchmarks/test_overlay_read.py -s -``` - -Size/byte metrics are attached with `record_property` (visible in `--junitxml` output) -and printed with `-s`; read/scan timings use pytest-benchmark (`--benchmark-json`). - ## Investigating memory use for a particular benchmark To investigate memory use for a particular benchmark, you can use the `bytehound` library. diff --git a/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py b/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py index 8339635c8df..0303e41c7a3 100644 --- a/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py +++ b/python/python/ci_benchmarks/benchmarks/test_overlay_manifest.py @@ -48,8 +48,3 @@ def test_overlay_manifest_size( record_property("manifest_bytes", total) record_property("manifest_growth_bytes", growth) record_property("bytes_per_overlay", per_overlay) - print( - f"\noverlays={num_overlays:>3} {fraction:>5.0%} {pattern:<10} " - f"manifest={total:>8}B growth={growth:>8}B " - f"per_overlay={per_overlay:>8.0f}B" - ) diff --git a/python/python/ci_benchmarks/benchmarks/test_overlay_read.py b/python/python/ci_benchmarks/benchmarks/test_overlay_read.py index edafc6813f4..27038d105ce 100644 --- a/python/python/ci_benchmarks/benchmarks/test_overlay_read.py +++ b/python/python/ci_benchmarks/benchmarks/test_overlay_read.py @@ -8,7 +8,7 @@ - the number of overlay layers stacked on a fragment (compaction payoff), - coverage fraction (how many cells are overlaid), - fragmentation (contiguous run vs. strided -> pages touched), and -- data type (int32 vs. a fixed-size-list embedding). +- value width (a 4-byte int32 vs. a wide fixed-size-list embedding). Wall time is measured warm via pytest-benchmark; read IO (bytes + IOPS) is measured once cold, after dropping the page cache, via ``io_stats_incremental``. @@ -22,7 +22,6 @@ from ci_benchmarks.utils import wipe_os_cache NUM_ROWS = 1_000_000 -NUM_ROWS_EMBEDDING = 100_000 ROWS_PER_FILE = NUM_ROWS # single fragment: isolates overlay-layer scaling TAKE_ROWS = 100 @@ -72,7 +71,6 @@ def work(): read_bytes, read_iops = _measure_cold_io(ds, base, work) record_property("cold_read_bytes", read_bytes) record_property("cold_read_iops", read_iops) - print(f"\ncold read: {read_bytes}B over {read_iops} iops") benchmark(work) @@ -104,18 +102,6 @@ def test_overlay_read_scaling( _run_read(benchmark, record_property, base, ds, workload, NUM_ROWS) -@pytest.mark.parametrize("workload", ["take", "scan"]) -@pytest.mark.parametrize("dtype", ["int32", "embedding"]) -def test_overlay_read_dtype(benchmark, tmp_path, record_property, workload, dtype): - num_rows = NUM_ROWS_EMBEDDING if dtype == "embedding" else NUM_ROWS - base = str(tmp_path / "ds") - ds = make_base_dataset(base, num_rows, num_rows, dtype, "2.1") - base_val = _covered_value(ds) - ds = commit_overlay_layers(ds, 4, 0.01, "stride", dtype) - assert _covered_value(ds) != base_val, "overlay not visible on read" - _run_read(benchmark, record_property, base, ds, workload, num_rows) - - # Mirror test_overlay_read_scaling but on a wide 3072-d embedding column, so the # take/scan-vs-layers story can be read for a fat value column rather than a # 4-byte one. Pinned to the 2.1 format (v2.0 layer scaling is already covered by From 86ed7836eb262a2f0299fc4961542c0c2153b42d Mon Sep 17 00:00:00 2001 From: Will Jones Date: Thu, 23 Jul 2026 10:19:40 -0700 Subject: [PATCH 6/8] test(python): remove overlay write-cost benchmark The write-cost comparison (overlay vs. update / merge_insert / full rewrite) is a useful one-off measurement but not something we need to keep re-running in CI for regressions. Remove it, along with the helpers it exclusively used (file_sizes, written_breakdown, proc_io_counters, and the make_base_dataset payload column). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../benchmarks/test_overlay_write.py | 163 ------------------ python/python/ci_benchmarks/overlays.py | 63 +------ 2 files changed, 2 insertions(+), 224 deletions(-) delete mode 100644 python/python/ci_benchmarks/benchmarks/test_overlay_write.py diff --git a/python/python/ci_benchmarks/benchmarks/test_overlay_write.py b/python/python/ci_benchmarks/benchmarks/test_overlay_write.py deleted file mode 100644 index 0a54c8a3ac7..00000000000 --- a/python/python/ci_benchmarks/benchmarks/test_overlay_write.py +++ /dev/null @@ -1,163 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright The Lance Authors - -"""Benchmark #2: cost of updating a single column for 1% of rows. - -Compares writing an overlay against the existing rewrite-based mechanisms -(``update``, ``merge_insert``, full overwrite). - -The rewrite mechanisms are delete-and-reinsert: they re-encode the changed rows -*across all columns* into a new fragment and add a deletion vector to each -affected fragment. An overlay instead persists only the changed column's cells -plus a coverage bitmap in the manifest. The crossover therefore depends on row -width, so the benchmark sweeps a ``width`` axis: - -- ``narrow``: id + val only. Rewriting whole rows is nearly free, so an overlay - has no byte advantage and its manifest/bitmap overhead can make it larger. -- ``wide``: id + val + a 64-d float32 payload. Rewrites re-encode the payload for - every changed row; the overlay still writes only ``val``. - -Persisted bytes are split into data (under ``data/``) and metadata (manifests, -transactions, deletion vectors), because the two families spend their bytes -differently. ``read_bytes`` (process-wide read syscalls from /proc/self/io) and -wall time are reported alongside: the rewrite mechanisms must read the affected -rows to re-encode them, which both costs IO and explains much of their extra -time, while an overlay reads essentially nothing. - -NOTE: the overlay arm is hand-rolled (no ``update``-via-overlay path exists yet) -and, unlike the other arms, does not read the base column first -- it writes -fresh values for the covered cells. That no-read is the win this read metric -exposes. -""" - -from time import perf_counter - -import lance -import numpy as np -import pyarrow as pa -import pytest -from ci_benchmarks.overlays import ( - commit_overlay_layers, - file_sizes, - make_base_dataset, - proc_io_counters, - written_breakdown, -) - -NUM_ROWS = 1_000_000 -ROWS_PER_FILE = 100_000 # 10 fragments; the strided selection hits every one -UPDATE_FRACTION = 0.01 -STEP = int(1 / UPDATE_FRACTION) # select id % STEP == 0 -WIDE_DIM = 64 - -APPROACHES = [ - "update", - "merge_insert", - "merge_insert_indexed", - "overlay", - "full_rewrite", -] - - -def _selected_ids() -> np.ndarray: - return np.arange(0, NUM_ROWS, STEP, dtype=np.int64) - - -def _payload(n: int, rng: np.random.Generator) -> pa.Array: - flat = rng.random(n * WIDE_DIM, dtype=np.float32) - return pa.FixedSizeListArray.from_arrays(pa.array(flat), WIDE_DIM) - - -def _new_rows(ids: np.ndarray, payload_dim: int) -> pa.Table: - """Full replacement rows for ``merge_insert`` (a row-oriented op cannot - target a single column, so it brings whole rows).""" - rng = np.random.default_rng(1) - cols = { - "id": pa.array(ids), - "val": pa.array(rng.integers(0, 1 << 30, len(ids), np.int32)), - } - if payload_dim: - cols["payload"] = _payload(len(ids), rng) - return pa.table(cols) - - -@pytest.mark.parametrize("width", ["narrow", "wide"]) -@pytest.mark.parametrize("approach", APPROACHES) -def test_overlay_write_cost(tmp_path, record_property, approach, width): - payload_dim = WIDE_DIM if width == "wide" else 0 - base = str(tmp_path / "ds") - ds = make_base_dataset( - base, NUM_ROWS, ROWS_PER_FILE, "int32", "2.1", payload_dim=payload_dim - ) - ids = _selected_ids() - sel, unsel = int(ids[0]), int(ids[0]) + 1 - before = { - r["id"]: r["val"] - for r in ds.to_table( - columns=["id", "val"], filter=f"id IN ({sel},{unsel})" - ).to_pylist() - } - - if approach == "merge_insert_indexed": - ds.create_scalar_index("id", "BTREE") # setup, not measured - - def run(): - nonlocal ds - if approach == "update": - ds.update({"val": "val + 1"}, where=f"id % {STEP} = 0") - elif approach in ("merge_insert", "merge_insert_indexed"): - ds.merge_insert("id").when_matched_update_all().execute( - _new_rows(ids, payload_dim) - ) - elif approach == "overlay": - ds = commit_overlay_layers(ds, 1, UPDATE_FRACTION, "stride", "int32") - elif approach == "full_rewrite": - table = ds.to_table() - val = table.column("val").to_numpy(zero_copy_only=False).copy() - val[ids] = _new_rows(ids, 0).column("val").to_numpy() - cols = {"id": table.column("id"), "val": pa.array(val)} - if payload_dim: - cols["payload"] = table.column("payload") - ds = lance.write_dataset( - pa.table(cols), base, mode="overwrite", data_storage_version="2.1" - ) - else: - raise ValueError(approach) - - snap_before = file_sizes(base) - io_before = proc_io_counters() - start = perf_counter() - run() - elapsed = perf_counter() - start - io_after = proc_io_counters() - snap_after = file_sizes(base) - - # Guard against silently measuring a no-op: the selected row must change and - # the unselected row must not. - ds = lance.dataset(base) - after = { - r["id"]: r["val"] - for r in ds.to_table( - columns=["id", "val"], filter=f"id IN ({sel},{unsel})" - ).to_pylist() - } - assert after[sel] != before[sel], f"{approach}/{width}: selected row unchanged" - assert after[unsel] == before[unsel], f"{approach}/{width}: unselected row changed" - - data_bytes, meta_bytes = written_breakdown(snap_before, snap_after) - # rchar/wchar (syscall bytes) are cache-independent; the base is warm from - # the build, so physical read_bytes would under-report what was read. - read_bytes = ( - io_after["rchar"] - io_before["rchar"] if io_before and io_after else -1 - ) - - record_property("data_bytes", data_bytes) - record_property("metadata_bytes", meta_bytes) - record_property("persisted_bytes", data_bytes + meta_bytes) - record_property("read_bytes", read_bytes) - record_property("seconds", elapsed) - print( - f"\n{approach:<22} {width:<7} read={read_bytes:>11}B " - f"data={data_bytes:>10}B meta={meta_bytes:>9}B " - f"persisted={data_bytes + meta_bytes:>10}B time={elapsed:>7.3f}s" - ) diff --git a/python/python/ci_benchmarks/overlays.py b/python/python/ci_benchmarks/overlays.py index 523a17ed613..13562fa13f9 100644 --- a/python/python/ci_benchmarks/overlays.py +++ b/python/python/ci_benchmarks/overlays.py @@ -16,7 +16,7 @@ # enable it here, before lance is imported, or reading overlay datasets fails. os.environ.setdefault("LANCE_ENABLE_UNSTABLE_DATA_OVERLAY_FILES", "1") -from typing import List, Optional, Tuple # noqa: E402 +from typing import List # noqa: E402 from urllib.parse import urlparse # noqa: E402 import lance # noqa: E402 @@ -57,7 +57,6 @@ def make_base_dataset( rows_per_file: int, dtype: str, version: str, - payload_dim: int = 0, embedding_dim: int = EMBEDDING_DIM, ) -> lance.LanceDataset: """Create a base dataset with an ``id`` key and a ``val`` payload column. @@ -65,15 +64,10 @@ def make_base_dataset( ``val`` is the column overlays target. ``rows_per_file`` controls the number of fragments (``num_rows // rows_per_file``). ``embedding_dim`` sets the width of the ``val`` column when ``dtype`` is ``embedding`` (e.g. 3072 for a wide - embedding). When ``payload_dim`` > 0 an extra ``payload`` fixed-size-list - float32 column of that width is added, making rows wider so the cost of - rewriting whole rows (update / merge_insert) can be compared against - overlaying just ``val``. + embedding). """ vtype = _value_type(dtype, embedding_dim) fields = {"id": pa.int64(), "val": vtype} - if payload_dim: - fields["payload"] = pa.list_(pa.float32(), payload_dim) schema = pa.schema(fields) rng = np.random.default_rng(0) @@ -83,11 +77,6 @@ def batches(): n = min(rows_per_file, num_rows - written) ids = pa.array(range(written, written + n), pa.int64()) cols = [ids, _gen_values(dtype, n, rng, embedding_dim)] - if payload_dim: - flat = rng.random(n * payload_dim, dtype=np.float32) - cols.append( - pa.FixedSizeListArray.from_arrays(pa.array(flat), payload_dim) - ) yield pa.record_batch(cols, schema=schema) written += n @@ -182,54 +171,6 @@ def _local_path(ds: lance.LanceDataset) -> str: return parsed.path if parsed.scheme == "file" else ds.uri -def proc_io_counters() -> Optional[dict]: - """Process-wide /proc/self/io counters, or None if unavailable. - - Captures IO regardless of which object store issued it, which is what makes - it the right tool for comparing mechanisms whose IO does not all flow through - a single dataset handle. Use ``rchar``/``wchar`` (bytes moved by read/write - syscalls) rather than ``read_bytes``/``write_bytes`` (physical storage IO): - freshly written data sits in the page cache, so physical reads under-report - what an operation actually read. - """ - try: - stats = {} - with open("/proc/self/io") as f: - for line in f: - key, _, value = line.partition(":") - stats[key.strip()] = int(value) - return stats - except OSError: - return None - - -def file_sizes(path: str) -> dict: - """Map of relative-path -> size in bytes for every file under ``path``.""" - out = {} - for root, _, files in os.walk(path): - for name in files: - fp = os.path.join(root, name) - if os.path.isfile(fp): - out[os.path.relpath(fp, path)] = os.path.getsize(fp) - return out - - -def written_breakdown(before: dict, after: dict) -> Tuple[int, int]: - """Split newly written/grown bytes (``after`` vs ``before`` snapshots) into - (data_bytes, metadata_bytes). Data is everything under ``data/``; metadata is - manifests, transactions, and deletion files.""" - data = meta = 0 - for path, size in after.items(): - delta = size - before.get(path, 0) - if delta <= 0: - continue - if path.startswith("data" + os.sep): - data += delta - else: - meta += delta - return data, meta - - def manifest_size(ds: lance.LanceDataset) -> int: """Size in bytes of the manifest for the dataset's current version.""" # Manifests are named `{u64::MAX - version}.manifest` so that a plain From ed641184c40f61bcc6042816e4dc470c2e94af55 Mon Sep 17 00:00:00 2001 From: Will Jones Date: Thu, 23 Jul 2026 11:16:01 -0700 Subject: [PATCH 7/8] feat(python): add row-count batching to rand_batches The datagen binding could only size generated batches by bytes. The underlying lance-datagen builder already supports an exact row count (into_reader_rows), so expose it: rand_batches gains a rows_per_batch option, mutually exclusive with batch_size_bytes. This lets callers request an exact number of rows per batch rather than approximating via a byte budget. Co-Authored-By: Claude Opus 4.8 (1M context) --- python/python/lance/_datagen.py | 5 +++- python/src/datagen.rs | 46 ++++++++++++++++++++++++--------- 2 files changed, 38 insertions(+), 13 deletions(-) diff --git a/python/python/lance/_datagen.py b/python/python/lance/_datagen.py index b156066eca6..45826503c5b 100644 --- a/python/python/lance/_datagen.py +++ b/python/python/lance/_datagen.py @@ -21,10 +21,13 @@ def rand_batches( *, num_batches: Optional[int] = None, batch_size_bytes: Optional[int] = None, + rows_per_batch: Optional[int] = None, ): if not datagen.is_datagen_supported(): raise NotImplementedError( "This version of lance was not built with the datagen feature" ) - batch_iter = datagen.rand_batches(schema, num_batches, batch_size_bytes) + batch_iter = datagen.rand_batches( + schema, num_batches, batch_size_bytes, rows_per_batch + ) return pa.RecordBatchReader.from_batches(schema, batch_iter) diff --git a/python/src/datagen.rs b/python/src/datagen.rs index 8b046c37f24..194ba3d6a73 100644 --- a/python/src/datagen.rs +++ b/python/src/datagen.rs @@ -1,7 +1,7 @@ use arrow::pyarrow::PyArrowType; -use arrow_array::RecordBatch; +use arrow_array::{RecordBatch, RecordBatchReader}; use arrow_schema::Schema; -use lance_datagen::{BatchCount, ByteCount}; +use lance_datagen::{BatchCount, ByteCount, RowCount}; use pyo3::{ Bound, PyResult, Python, pyfunction, types::{PyModule, PyModuleMethods}, @@ -16,22 +16,44 @@ pub fn is_datagen_supported() -> bool { true } +/// Generate `batch_count` batches of random data for `schema`. +/// +/// Batch size is set either by `rows_in_batch` (exact rows per batch) or +/// `bytes_in_batch` (approximate bytes per batch); the two are mutually +/// exclusive. When neither is given the byte-based default is used. #[pyfunction] -#[pyo3(signature=(schema, batch_count=None, bytes_in_batch=None))] +#[pyo3(signature=(schema, batch_count=None, bytes_in_batch=None, rows_in_batch=None))] pub fn rand_batches( schema: PyArrowType, batch_count: Option, bytes_in_batch: Option, + rows_in_batch: Option, ) -> PyResult>> { - lance_datagen::rand(&schema.0) - .into_reader_bytes( - ByteCount::from(bytes_in_batch.unwrap_or(DEFAULT_BATCH_SIZE_BYTES)), - BatchCount::from(batch_count.unwrap_or(DEFAULT_BATCH_COUNT)), - lance_datagen::RoundingBehavior::RoundUp, - ) - .map_err(|e| { - pyo3::exceptions::PyValueError::new_err(format!("Failed to generate batches: {}", e)) - })? + if rows_in_batch.is_some() && bytes_in_batch.is_some() { + return Err(pyo3::exceptions::PyValueError::new_err( + "rows_in_batch and bytes_in_batch are mutually exclusive", + )); + } + let builder = lance_datagen::rand(&schema.0); + let batch_count = BatchCount::from(batch_count.unwrap_or(DEFAULT_BATCH_COUNT)); + let reader: Box = match rows_in_batch { + Some(rows) => Box::new(builder.into_reader_rows(RowCount::from(rows), batch_count)), + None => Box::new( + builder + .into_reader_bytes( + ByteCount::from(bytes_in_batch.unwrap_or(DEFAULT_BATCH_SIZE_BYTES)), + batch_count, + lance_datagen::RoundingBehavior::RoundUp, + ) + .map_err(|e| { + pyo3::exceptions::PyValueError::new_err(format!( + "Failed to generate batches: {}", + e + )) + })?, + ), + }; + reader .map(|item| { item.map(PyArrowType::from).map_err(|e| { pyo3::exceptions::PyValueError::new_err(format!("Failed to generate batch: {}", e)) From 1be2bec63bcf1488075ee9c0254c0119e6681795 Mon Sep 17 00:00:00 2001 From: Will Jones Date: Thu, 23 Jul 2026 11:16:05 -0700 Subject: [PATCH 8/8] test(python): generate overlay base datasets via rand_batches Replace the hand-rolled base-dataset generator in make_base_dataset with lance._datagen.rand_batches (row-count mode), matching the other ci_benchmarks suites. One batch per fragment; num_rows must be a multiple of rows_per_file. Overlay values keep their own generator so each layer stays distinct from the base (the read-visibility guard depends on that). Co-Authored-By: Claude Opus 4.8 (1M context) --- python/python/ci_benchmarks/overlays.py | 35 +++++++++++-------------- 1 file changed, 15 insertions(+), 20 deletions(-) diff --git a/python/python/ci_benchmarks/overlays.py b/python/python/ci_benchmarks/overlays.py index 13562fa13f9..c2610e36d2e 100644 --- a/python/python/ci_benchmarks/overlays.py +++ b/python/python/ci_benchmarks/overlays.py @@ -22,6 +22,7 @@ import lance # noqa: E402 import numpy as np # noqa: E402 import pyarrow as pa # noqa: E402 +from lance._datagen import rand_batches # noqa: E402 from lance.file import LanceFileWriter # noqa: E402 # Default width for the ``embedding`` dtype. Individual benchmarks override it @@ -59,32 +60,26 @@ def make_base_dataset( version: str, embedding_dim: int = EMBEDDING_DIM, ) -> lance.LanceDataset: - """Create a base dataset with an ``id`` key and a ``val`` payload column. + """Create a base dataset with an ``id`` column and a ``val`` column. ``val`` is the column overlays target. ``rows_per_file`` controls the number - of fragments (``num_rows // rows_per_file``). ``embedding_dim`` sets the width - of the ``val`` column when ``dtype`` is ``embedding`` (e.g. 3072 for a wide - embedding). + of fragments (``num_rows // rows_per_file``), which must divide ``num_rows``. + ``embedding_dim`` sets the width of the ``val`` column when ``dtype`` is + ``embedding`` (e.g. 3072 for a wide embedding). """ - vtype = _value_type(dtype, embedding_dim) - fields = {"id": pa.int64(), "val": vtype} - schema = pa.schema(fields) - rng = np.random.default_rng(0) - - def batches(): - written = 0 - while written < num_rows: - n = min(rows_per_file, num_rows - written) - ids = pa.array(range(written, written + n), pa.int64()) - cols = [ids, _gen_values(dtype, n, rng, embedding_dim)] - yield pa.record_batch(cols, schema=schema) - written += n - - reader = pa.RecordBatchReader.from_batches(schema, batches()) + if num_rows % rows_per_file: + raise ValueError( + f"num_rows ({num_rows}) must be a multiple of rows_per_file " + f"({rows_per_file})" + ) + schema = pa.schema({"id": pa.int64(), "val": _value_type(dtype, embedding_dim)}) + # One fragment per batch; lance-datagen fills both columns with random data. + reader = rand_batches( + schema, num_batches=num_rows // rows_per_file, rows_per_batch=rows_per_file + ) return lance.write_dataset( reader, base_path, - schema=schema, max_rows_per_file=rows_per_file, data_storage_version=version, )