merge_insert should provide a configurable RAM limit

[westonpace]

merge_insert uses a datafusion join internally. Since we do not provide a RAM limit to datafusion it uses a lot of RAM by default to run the JOIN. We should restrain this and make it configurable. This will force datafusion to store to disk during the join which should save RAM at the cost of merge_insert performance.

[yeung108]

I hit this while debugging an OOMKilled batch ingestion job (chunked upserts into LanceDB on GCS, lancedb==0.33.0, pylance==7.0.0), and put together a controlled, isolated reproduction that I think adds concrete evidence to this issue.

Repro: single merge_insert() call's RSS cost scales with existing target table size, not incoming batch size (fixed at 500 rows throughout). Each row measured in a fresh subprocess for isolation, with a throwaway warm-up merge_insert() first so Lance/DataFusion/tokio runtime init cost doesn't pollute the measurement:

"""pip install lancedb pyarrow psutil"""
import gc, os, random, string, sys, tempfile
import lancedb, pyarrow as pa, psutil

proc = psutil.Process(os.getpid())
def rss_kb():
    gc.collect()
    return proc.memory_info().rss // 1024

SCHEMA = pa.schema([pa.field("id", pa.string()), pa.field("blob", pa.string())])
def rows(n, offset):
    return [{"id": f"{offset+i}", "blob": "".join(random.choices(string.ascii_letters, k=200))} for i in range(n)]

target_size = int(sys.argv[1])

# Warm-up: pay one-time runtime init cost on a throwaway connection/table.
warm_db = lancedb.connect(tempfile.mkdtemp())
warm_table = warm_db.create_table("W", data=pa.Table.from_pylist(rows(1, -1), SCHEMA), mode="create")
warm_table.merge_insert("id").when_matched_update_all().when_not_matched_insert_all().execute(
    pa.Table.from_pylist(rows(1, -2), SCHEMA)
)
del warm_table, warm_db
gc.collect()

# Build target table of the given size via add() (not merge_insert, so setup
# cost doesn't contaminate the single call being measured).
db = lancedb.connect(tempfile.mkdtemp())
table = db.create_table("T", data=pa.Table.from_pylist(rows(min(target_size, 5000) or 1, -1), SCHEMA), mode="create")
remaining = target_size - table.count_rows()
offset = 0
while remaining > 0:
    batch = min(remaining, 5000)
    table.add(pa.Table.from_pylist(rows(batch, offset), SCHEMA))
    offset += batch
    remaining -= batch

incoming = pa.Table.from_pylist(rows(500, offset=10**9), SCHEMA)  # non-matching ids
before = rss_kb()
table.merge_insert("id").when_matched_update_all().when_not_matched_insert_all().execute(incoming)
after = rss_kb()
print(f"target_rows={table.count_rows()-500:>8} single_call_delta_kb={after-before:>8,}")

Run for increasing target sizes (for n in 0 5000 20000 50000 100000 200000 400000; do python3 repro.py $n; done):

Target table rows	Single merge_insert() call RSS delta
1	2.4 MB
5,000	0.3 MB
20,000	0.6 MB
50,000	2.0 MB
100,000	5.9 MB
200,000	16.4 MB
400,000	30.7 MB

The incoming batch is fixed at 500 rows the whole time, so the only thing scaling here is the target table size — consistent with the unbounded DataFusion join described above. I also confirmed this isn't just a one-off transient peak: repeatedly calling merge_insert() against the same growing table (rather than measuring one call in isolation) shows RSS climbing and not coming back down between calls (+52MB current RSS via psutil, not just ru_maxrss, over 200 calls growing a table to 100k rows), while the identical number of merge_insert() calls each against a fresh near-empty table shows no growth at all (-0.5MB) — ruling out simple handle/connection caching as the cause and pointing squarely at the join itself.

This is a real operational hazard for streaming/batch ingestion patterns (chunked upserts into a table that grows over the job's lifetime) — it silently gets worse the longer the job runs, and eventually OOMs.

[yeung108]

Follow-up to my comment above: adding a scalar index on the merge key does not fix this — RSS still grows unboundedly across a long sequence of merge_insert() calls, even with periodic optimize() calls to fold the unindexed backlog into the index (the maintenance cadence the docs recommend).

Environment: `lancedb==0.33.0`, `pylance==7.0.0`, macOS arm64.

"""pip install lancedb pyarrow psutil"""
import argparse, gc, os, random, string, tempfile
import lancedb, psutil, pyarrow as pa

SCHEMA = pa.schema([pa.field("id", pa.string()), pa.field("blob", pa.string())])

def _rows(n, offset, blob_bytes):
    return [{"id": str(offset + i), "blob": "".join(random.choices(string.ascii_letters, k=blob_bytes))} for i in range(n)]

def _dir_size_bytes(path):
    return sum(os.path.getsize(os.path.join(root, name)) for root, _, files in os.walk(path) for name in files)

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--target-size", type=int, default=5_000)
    parser.add_argument("--iterations", type=int, default=200)
    parser.add_argument("--chunk-size", type=int, default=500)
    parser.add_argument("--optimize-every", type=int, default=2_000)
    parser.add_argument("--blob-bytes", type=int, default=512)
    parser.add_argument("--sample-every", type=int, default=10)
    args = parser.parse_args()

    proc = psutil.Process(os.getpid())
    def rss_mb():
        gc.collect()
        return proc.memory_info().rss / 1e6

    db_path = tempfile.mkdtemp(prefix="lance_issue_1983_")
    db = lancedb.connect(db_path)

    # Seed the target table and index its merge key before measuring.
    table = db.create_table("t", data=pa.Table.from_pylist(_rows(min(args.target_size, 5000) or 1, 0, args.blob_bytes), SCHEMA), mode="create")
    offset = table.count_rows()
    while offset < args.target_size:
        batch = min(args.target_size - offset, 5000)
        table.add(pa.Table.from_pylist(_rows(batch, offset, args.blob_bytes), SCHEMA))
        offset += batch
    table.create_scalar_index("id", index_type="BTREE")

    print("# iter  total_rows     rss_mb  rss_growth_mb  pool_bytes_allocated")
    baseline = rss_mb()
    since_optimize = 0
    offset = 10**9  # disjoint from seeded ids -> every merge_insert is a pure insert
    for i in range(1, args.iterations + 1):
        incoming = pa.Table.from_pylist(_rows(args.chunk_size, offset, args.blob_bytes), SCHEMA)
        offset += args.chunk_size
        table.merge_insert("id").when_matched_update_all().when_not_matched_insert_all().execute(incoming)
        since_optimize += args.chunk_size
        if since_optimize >= args.optimize_every:
            table.optimize()
            since_optimize = 0
        if i % args.sample_every == 0 or i in (1, args.iterations):
            rss = rss_mb()
            print(f"{i:>6}  {table.count_rows():>10}  {rss:>9.1f}  {rss - baseline:>+13.1f}  {pa.default_memory_pool().bytes_allocated():>20}")

    disk_mb = _dir_size_bytes(db_path) / 1e6
    print(f"\non-disk dataset size: {disk_mb:.1f} MB   |   RSS growth: {rss_mb() - baseline:.1f} MB   |   pool bytes allocated: {pa.default_memory_pool().bytes_allocated()}")

if __name__ == "__main__":
    main()

Output (python3 repro.py, defaults: target=5000 iterations=200 chunk=500 optimize_every=2000 blob_bytes=512):

# iter  total_rows     rss_mb  rss_growth_mb  pool_bytes_allocated
     1        5500      202.5           +4.9                265152
    50       30000      387.4         +189.7                265152
   100       55000      485.1         +287.5                265152
   150       80000      630.5         +432.9                265152
   200      105000      662.0         +464.4                265152

on-disk dataset size: 1502.0 MB   |   RSS growth: 464.4 MB   |   pool bytes allocated: 265152

A few things this rules out:

• Not the incoming batch: it's a fixed 500 rows/call the whole time.
• Not PyArrow's own memory pool: pa.default_memory_pool().bytes_allocated() is flat at a constant 265152 bytes for the entire 200-call run.
• Not an unindexed join: the id column carries a BTREE scalar index for the whole measured loop, and table.optimize() runs every 2000 rows (matching Lance's own guidance) to fold any unindexed backlog into it.

RSS still grows ~9x larger than the actual payload bytes written in the loop (200 x 500 x 512 bytes = 51.2 MB written vs. 464 MB of RSS growth), and never comes back down between calls. The on-disk dataset directory also balloons far past what the live row count implies (1.5 GB on disk for a table that never holds more than ~105k narrow rows), consistent with fragment/manifest versions from each merge_insert()/optimize() call being retained rather than reclaimed.

This makes "index the merge key" insufficient as a workaround for long-running/streaming upsert workloads — the RAM-limit ask in this issue (or a spill-to-disk option for the join, or more aggressive fragment/version reclamation) looks like the right fix rather than relying on indexing alone.

[yeung108]

Follow-up with jemalloc-instrumented live-heap measurement — this looks like a genuine leak in Lance's write path, not something specific to merge_insert's join, and not explained by version/fragment retention.

My previous comment's disk-bloat observation was a confound: lancedb's optimize() defaults cleanup_older_than to 7 days, so a seconds-long test loop reclaims nothing. Re-running with cleanup_older_than=timedelta(0) keeps disk usage flat (~2.5–3 MB) and fragment/version counts bounded (1–7) for the whole run, but RSS growth is essentially unchanged (489 MB vs. 484 MB over 200 merge_insert() calls growing a table from 5k→105k rows) — so on-disk retention isn't the driver of the RSS growth, even though it's a real, separate issue worth fixing on its own.

To find the actual driver I ran under Linux with LD_PRELOAD=libjemalloc.so.2 (Lance uses the system allocator with no #[global_allocator] override, so this captures every Rust allocation Lance makes) and measured stats.allocated (jemalloc's live/in-use heap — unaffected by allocator page-retention settings) alongside /proc/self/smaps_rollup's anonymous-vs-file-backed RSS split. Setup: table held at a constant 5,000 live rows (delete the oldest chunk after each insert), cleanup_older_than=0 on every optimize, dirty_decay_ms:0,muzzy_decay_ms:0 (jemalloc's own dirty-page retention disabled), 1,500 write calls, periodic forced arena.purge.

Writer	stats.allocated growth (live heap)	RSS growth	mmap (file-backed RSS)	Forced allocator purge recovered
table.add() — no merge_insert, no join, no index	+77.8 MB	+161.0 MB	flat after warm-up	0.0 MB, every time
merge_insert() + BTREE index on the join key	+117.8 MB	+211.2 MB	flat after warm-up	0.0 MB, every time

This rules out the alternative explanations simultaneously: mmap is flat (not fragment/page-cache growth), disk/versions are bounded (not un-reclaimed history), and a forced allocator purge recovers nothing (not allocator retention — stats.allocated only decreases on free, independent of decay/purge policy). Live heap growing under a constant table size, with disk/fragments/versions provably not growing, unrecoverable by purge, looks like a leak.

Notably, table.add() alone leaks — merge_insert isn't required to reproduce this, though it leaks ~50% faster than plain add() (consistent with the join/index-lookup path allocating additional structures on top of a pre-existing leak in the base write/commit path). That suggests a RAM limit on the join (this issue's original ask) would only bound the extra cost merge_insert adds — a long-running writer using plain add()/append would still eventually OOM.

Repro script:

"""pip install lancedb pyarrow psutil

Arbiter for whether the RSS growth shown by repeated merge_insert()/add() calls is a genuine
Rust-side heap leak, or one of three non-leak causes that also produce climbing RSS:

    1. True live-heap leak    -- Rust allocations grow and are never freed.
    2. Allocator retention    -- freed memory kept in malloc arenas, not returned to the OS.
    3. mmap of on-disk data   -- growing fragment/version files are file-backed resident pages.
    4. Un-reclaimed versions  -- optimize()'s default cleanup_older_than is 7 days, so a
                                 seconds-long loop reclaims nothing; disk + scan working-set grow.

psutil RSS cannot distinguish these. This script adds three measurements that can:

  - jemalloc_stats(): reads `stats.allocated` via ctypes mallctl. Lance ships with the *system*
    allocator (no #[global_allocator] override), so on Linux with `LD_PRELOAD=libjemalloc.so.2`
    this captures the live heap of every Rust allocation Lance makes -- the one number that can't
    be inflated by retention or mmap. None on macOS / without LD_PRELOAD.
  - rss_split_kb(): parses /proc/self/smaps_rollup into anonymous (heap) vs. file-backed (mmap)
    resident KB. None on macOS.
  - purge(): forces the allocator to return freed pages to the OS (jemalloc arena purge, or glibc
    malloc_trim). If RSS drops after purge, the pre-purge growth was retention, not a leak.

Four confound-removal knobs isolate one hypothesis at a time:

    --cleanup zero    passes cleanup_older_than=timedelta(0) to optimize(), removing cause 4.
    --writer add      skips merge_insert's join entirely, isolating whether the join is the driver.
    --steady          deletes the oldest --chunk-size rows after every write, holding live row
                       count constant so growth can't be explained by "the dataset got bigger."
    --no-index        matches the original unindexed repro (default is indexed, matching the
                       stricter setup from my prior comment).

The decisive cell is `--writer merge_insert --cleanup zero --steady` on Linux with jemalloc:
constant live data, no version retention, purge-verified allocator state. If je_alloc_mb still
climbs there, it's a genuine leak. If it's flat while rss_mb climbs, the growth is retention/mmap,
not a leak -- see run_leak_arbiter_container.sh (below) for the jemalloc-enabled container run.

Usage:
    python3 repro_merge_insert_leak_arbiter.py --writer merge_insert --cleanup default
    python3 repro_merge_insert_leak_arbiter.py --writer merge_insert --cleanup zero
    python3 repro_merge_insert_leak_arbiter.py --writer add          --cleanup zero
    python3 repro_merge_insert_leak_arbiter.py --writer merge_insert --cleanup zero --steady
"""

import argparse
import ctypes
import gc
import os
import random
import resource
import string
import sys
import tempfile
from datetime import timedelta

import lancedb
import psutil
import pyarrow as pa

proc = psutil.Process(os.getpid())

# jemalloc's reserved arena index meaning "all arenas" (jemalloc/internal/jemalloc_internal_types.h).
_MALLCTL_ARENAS_ALL = 4096


def _load_mallctl():
    # If the process was started with LD_PRELOAD=libjemalloc.so.2 (Linux only -- Debian/Ubuntu's
    # libjemalloc2 package is built with unprefixed symbol names specifically to support this),
    # jemalloc's malloc/free/mallctl preempt libc's in the global symbol scope, so dlsym via
    # CDLL(None) finds "mallctl" even though we never linked against jemalloc directly. On glibc
    # without a jemalloc preload, or on macOS, this symbol doesn't exist and we fall back to None
    # everywhere below.
    try:
        libc = ctypes.CDLL(None)
        mallctl = libc.mallctl
    except (OSError, AttributeError):
        return None
    mallctl.restype = ctypes.c_int
    mallctl.argtypes = [
        ctypes.c_char_p,
        ctypes.c_void_p,
        ctypes.POINTER(ctypes.c_size_t),
        ctypes.c_void_p,
        ctypes.c_size_t,
    ]
    return mallctl


_MALLCTL = _load_mallctl()


def _mallctl_read_u64(name: bytes) -> int | None:
    if _MALLCTL is None:
        return None
    value = ctypes.c_uint64(0)
    size = ctypes.c_size_t(ctypes.sizeof(value))
    rc = _MALLCTL(name, ctypes.byref(value), ctypes.byref(size), None, 0)
    return value.value if rc == 0 else None


def jemalloc_stats() -> dict[str, int] | None:
    """Live jemalloc heap stats in bytes, or None if jemalloc isn't preloaded into this process."""
    if _MALLCTL is None:
        return None
    # "epoch" is a write-triggered mallctl: writing any u64 to it refreshes jemalloc's cached
    # stats before the stats.* reads below, otherwise they can return stale values.
    epoch = ctypes.c_uint64(1)
    size = ctypes.c_size_t(ctypes.sizeof(epoch))
    _MALLCTL(b"epoch", ctypes.byref(epoch), ctypes.byref(size), ctypes.byref(epoch), size)
    allocated = _mallctl_read_u64(b"stats.allocated")
    if allocated is None:
        return None
    return {
        "allocated": allocated,  # live heap: bytes actually in use by the application
        "resident": _mallctl_read_u64(b"stats.resident") or 0,
        "retained": _mallctl_read_u64(b"stats.retained") or 0,  # freed, held by jemalloc, not returned to OS
        "mapped": _mallctl_read_u64(b"stats.mapped") or 0,
    }


def purge() -> str | None:
    """Force the allocator to return freed pages to the OS. Returns which backend did it, if any."""
    if _MALLCTL is not None:
        rc = _MALLCTL(f"arena.{_MALLCTL_ARENAS_ALL}.purge".encode(), None, None, None, 0)
        if rc == 0:
            return "jemalloc"
    if sys.platform == "linux":
        try:
            ctypes.CDLL(None).malloc_trim(0)
            return "glibc"
        except (OSError, AttributeError):
            pass
    return None


def rss_kb() -> int:
    gc.collect()
    return proc.memory_info().rss // 1024


def peak_rss_kb() -> int:
    raw = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    return raw // 1024 if sys.platform == "darwin" else raw


def rss_split_kb() -> tuple[int, int] | tuple[None, None]:
    """Anonymous (heap-like) vs. file-backed (mmap-like) resident KB, from /proc. Linux only."""
    path = "/proc/self/smaps_rollup"
    if not os.path.exists(path):
        return None, None
    total_rss = anon = 0
    with open(path) as f:
        for line in f:
            if line.startswith("Rss:"):
                total_rss = int(line.split()[1])
            elif line.startswith("Anonymous:"):
                anon = int(line.split()[1])
    return anon, total_rss - anon


def _dir_size_bytes(path: str) -> int:
    return sum(os.path.getsize(os.path.join(root, name)) for root, _, files in os.walk(path) for name in files)


# id is int64 (not the string ids used by my earlier repro scripts) so --steady's window deletes
# can use a plain numeric threshold ("id < N") without needing zero-padded string ordering.
SCHEMA = pa.schema([pa.field("id", pa.int64()), pa.field("blob", pa.string())])


def rows(n: int, offset: int, blob_bytes: int) -> list[dict]:
    return [
        {"id": offset + i, "blob": "".join(random.choices(string.ascii_letters, k=blob_bytes))} for i in range(n)
    ]


def fill(table, n: int, offset: int, blob_bytes: int) -> int:
    remaining = n
    while remaining > 0:
        batch = min(remaining, 5000)
        table.add(pa.Table.from_pylist(rows(batch, offset, blob_bytes), SCHEMA))
        offset += batch
        remaining -= batch
    return offset


def unindexed_backlog(table) -> int:
    indices = table.list_indices()
    if not indices:
        return 0
    first = indices[0]
    name = first["name"] if isinstance(first, dict) else first.name
    stats = table.to_lance().stats.index_stats(name)
    return stats.get("num_unindexed_rows") or 0


def dataset_shape(table) -> tuple[int, int]:
    """(fragment_count, version_count) -- both should stay bounded if versions are truly reclaimed."""
    ds = table.to_lance()
    return len(ds.get_fragments()), len(list(ds.versions()))


def do_write(table, writer: str, incoming: pa.Table, insert_only: bool) -> None:
    if writer == "add":
        table.add(incoming)
        return
    builder = table.merge_insert("id")
    if not insert_only:
        builder = builder.when_matched_update_all()
    builder.when_not_matched_insert_all().execute(incoming)


def do_optimize(table, cleanup: str) -> None:
    if cleanup == "zero":
        table.optimize(cleanup_older_than=timedelta(seconds=0), delete_unverified=True)
    else:
        table.optimize()  # default cleanup_older_than=7 days -- reclaims nothing in a short-lived run


def _fmt(value: float | None, width: int = 9, prec: int = 1) -> str:
    return f"{value:>{width}.{prec}f}" if value is not None else f"{'n/a':>{width}}"


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--target-size", type=int, default=5_000, help="Rows in the target table before the loop.")
    parser.add_argument("--iterations", type=int, default=200, help="Number of write calls in the measured loop.")
    parser.add_argument("--chunk-size", type=int, default=500, help="Rows per write call.")
    parser.add_argument("--optimize-every", type=int, default=2_000, help="Call table.optimize() every N rows written.")
    parser.add_argument("--blob-bytes", type=int, default=512, help="Payload size per row.")
    parser.add_argument("--sample-every", type=int, default=10, help="Print a line every N iterations.")
    parser.add_argument(
        "--writer",
        choices=["merge_insert", "add"],
        default="merge_insert",
        help="'add' skips the join entirely -- isolates whether the join drives the growth.",
    )
    parser.add_argument(
        "--cleanup",
        choices=["default", "zero"],
        default="default",
        help="'zero' passes cleanup_older_than=timedelta(0) to optimize(), reclaiming old versions immediately "
        "instead of lancedb's 7-day default (which reclaims nothing in a short-lived run).",
    )
    parser.add_argument(
        "--steady",
        action="store_true",
        help="Delete the oldest --chunk-size rows after every write, holding live row count constant "
        "so growth can't be attributed to 'the dataset got bigger'.",
    )
    parser.add_argument(
        "--index", action=argparse.BooleanOptionalAction, default=True, help="Build a BTREE index on id (default on)."
    )
    parser.add_argument(
        "--insert-only", action="store_true", help="Skip when_matched_update_all() (merge_insert writer only)."
    )
    parser.add_argument(
        "--purge-every",
        type=int,
        default=0,
        help="Force-purge the allocator every N iterations and print RSS before/after (0 = never). "
        "If RSS drops after purge, the growth up to that point was retention, not a leak.",
    )
    args = parser.parse_args()

    je_probe = jemalloc_stats()
    print(
        f"# platform={sys.platform} jemalloc={'yes' if je_probe else 'no'} "
        f"smaps={'yes' if os.path.exists('/proc/self/smaps_rollup') else 'no'}",
        flush=True,
    )
    print(
        f"# writer={args.writer} cleanup={args.cleanup} steady={args.steady} index={args.index} "
        f"target={args.target_size} iterations={args.iterations} chunk={args.chunk_size} "
        f"optimize_every={args.optimize_every} blob_bytes={args.blob_bytes} insert_only={args.insert_only}",
        flush=True,
    )

    # Warm-up: pay one-time runtime init cost (tokio runtime, DataFusion session, etc.) on a
    # throwaway connection/table so it doesn't pollute the measured loop below.
    warm_db = lancedb.connect(tempfile.mkdtemp())
    warm = warm_db.create_table("W", data=pa.Table.from_pylist(rows(1, -1, 32), SCHEMA), mode="create")
    warm.merge_insert("id").when_matched_update_all().when_not_matched_insert_all().execute(
        pa.Table.from_pylist(rows(1, -2, 32), SCHEMA)
    )
    del warm, warm_db
    gc.collect()

    db_path = tempfile.mkdtemp(prefix="lance_leak_arbiter_")
    db = lancedb.connect(db_path)
    table = db.create_table(
        "T", data=pa.Table.from_pylist(rows(min(args.target_size, 5000) or 1, 0, args.blob_bytes), SCHEMA), mode="create"
    )
    fill(table, args.target_size - table.count_rows(), table.count_rows(), args.blob_bytes)
    if args.index:
        table.create_scalar_index("id", index_type="BTREE")

    header = (
        "# iter  live_rows  unindexed  disk_mb  frags  versions  "
        "rss_mb  rss_delta  anon_mb  file_mb  je_alloc_mb  je_resident_mb  je_retained_mb"
    )
    print(header, flush=True)

    baseline_rss = rss_kb() / 1024
    baseline_je = jemalloc_stats()
    next_id = table.count_rows()  # first id not yet used; keeps every write's ids disjoint from live data
    low = 0  # steady-mode sliding-window lower bound
    since_optimize = 0

    for i in range(1, args.iterations + 1):
        incoming = pa.Table.from_pylist(rows(args.chunk_size, next_id, args.blob_bytes), SCHEMA)
        next_id += args.chunk_size

        do_write(table, args.writer, incoming, args.insert_only)

        if args.steady:
            low_bound = low + args.chunk_size
            table.delete(f"id < {low_bound}")
            low = low_bound

        since_optimize += args.chunk_size
        if since_optimize >= args.optimize_every:
            do_optimize(table, args.cleanup)
            since_optimize = 0

        if args.purge_every and i % args.purge_every == 0:
            pre = rss_kb() / 1024
            backend = purge()
            post = rss_kb() / 1024
            if backend:
                print(f"    [purge:{backend}] rss {pre:.1f}MB -> {post:.1f}MB (freed {pre - post:+.1f}MB)", flush=True)

        if i % args.sample_every == 0 or i in (1, args.iterations):
            rss_mb = rss_kb() / 1024
            anon_kb, file_kb = rss_split_kb()
            anon_mb = anon_kb / 1024 if anon_kb is not None else None
            file_mb = file_kb / 1024 if file_kb is not None else None
            je = jemalloc_stats()
            je_alloc_mb = je["allocated"] / 1e6 if je else None
            je_resident_mb = je["resident"] / 1e6 if je else None
            je_retained_mb = je["retained"] / 1e6 if je else None
            frags, versions = dataset_shape(table)
            disk_mb = _dir_size_bytes(db_path) / 1e6
            print(
                f"{i:>6}  {table.count_rows():>9}  {unindexed_backlog(table):>9}  {disk_mb:>7.1f}  "
                f"{frags:>5}  {versions:>8}  {rss_mb:>6.1f}  {rss_mb - baseline_rss:>+9.1f}  "
                f"{_fmt(anon_mb, 7)}  {_fmt(file_mb, 7)}  {_fmt(je_alloc_mb, 11)}  "
                f"{_fmt(je_resident_mb, 14)}  {_fmt(je_retained_mb, 14)}",
                flush=True,
            )

    final_rss = rss_kb() / 1024
    final_je = jemalloc_stats()
    print(f"\n# summary: rss_growth_mb={final_rss - baseline_rss:.1f}", end="")
    if baseline_je and final_je:
        print(f"  je_alloc_growth_mb={(final_je['allocated'] - baseline_je['allocated']) / 1e6:.1f}", end="")
        print(f"  je_retained_mb={final_je['retained'] / 1e6:.1f}", end="")
    print(f"\n# db left at: {db_path}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

Container runner (Linux + jemalloc, since macOS has no LD_PRELOAD/`/proc` equivalent):

#!/usr/bin/env bash
# Runs repro_merge_insert_leak_arbiter.py in Linux with jemalloc LD_PRELOAD'd, so
# jemalloc_stats() reads a true live-heap number (Lance uses the system allocator,
# so LD_PRELOAD captures every allocation it makes). dirty_decay_ms:0,muzzy_decay_ms:0
# disables jemalloc's own page retention, so any RSS growth left over is either a
# real leak (stats.allocated climbing) or mmap of on-disk data, not retention.
#
# Usage: ./run_leak_arbiter_container.sh --writer merge_insert --cleanup zero --steady

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

docker run --rm --memory=2g --memory-swap=2g -v "$SCRIPT_DIR":/scripts:ro python:3.12-slim bash -c '
  set -euo pipefail
  apt-get update -qq && apt-get install -y -qq --no-install-recommends libjemalloc2 >/dev/null
  pip install --quiet lancedb==0.33.0 pylance==7.0.0 pyarrow psutil
  export LD_PRELOAD=$(find /usr/lib -name "libjemalloc.so.2" | head -1)
  export MALLOC_CONF="background_thread:true,dirty_decay_ms:0,muzzy_decay_ms:0"
  python3 /scripts/repro_merge_insert_leak_arbiter.py "$@"
' _ "$@"

[yeung108]

Root cause, traced into source, and a PR: #7718

Following up on my jemalloc-instrumented reproduction above (plain table.add() leaks too, not just merge_insert()'s join, ruling out allocator retention and on-disk version bloat) — I traced this into the actual Lance source and found the mechanism. Opening a PR against the real root cause rather than just the join's RAM usage, since the join turns out to only be part of the story.

The mechanism

Every commit inserts new entries into the session's metadata/index caches (rust/lance/src/session/caches.rs, rust/lance/src/session/index_caches.rs), keyed by dataset version:

ManifestKey { version }        // "manifest/{version}[/{e_tag}]"
TransactionKey { version }     // "txn/{version}"
RowAddrMaskKey { version, .. } // "row_addr_mask/{version}[/{hash}]"
RowIdIndexKey { version }      // "row_id_index/{version}"
IndexMetadataKey { version }   // bare version number, in the index cache

Nothing in the write, commit, or cleanup paths ever evicts these. A repo-wide search confirms the only call sites for the existing invalidate_prefix are the trait definition, its Moka implementation, and two test-only mock backends — never in commit.rs, cleanup.rs, or optimize.rs. So every commit permanently grows these caches, bounded only by DEFAULT_METADATA_CACHE_SIZE/DEFAULT_INDEX_CACHE_SIZE (1 GiB each) — a ceiling that's frequently above a container's actual memory limit, so the OOM killer fires long before Moka's own eviction ever would.

This maps exactly onto everything in the thread above:

Observation	Explanation
RSS/live-heap grows every call as the table grows	New version → new manifest/txn/row_addr_mask/row_id_index cache entries every commit; never evicted
Growth persists holding row count constant	Each write/delete cycle still creates new fragment UUIDs and a new version; old entries are orphaned, not invalidated
Scalar index + optimize() doesn't fix it	The index cache has the identical pattern (IndexMetadataKey{version}) — indexing just adds a second growing cache
cleanup_older_than=0 doesn't fix it	Cleanup deletes on-disk files; nothing ties that to in-memory cache invalidation — confirmed zero call sites
Forced jemalloc arena purge recovers 0.0 MB	These are legitimately live, referenced cache entries, not freed-but-retained allocator pages
Plain table.add() leaks too (just slower than merge_insert)	add() still creates new manifest/txn/fragment cache entries every commit; merge_insert()'s join additionally builds/caches row_id_index/row_addr_mask, which are bigger

So the RAM-limit ask in this issue's title is real, but it only bounds the extra memory merge_insert's join adds on top of a base write/commit-path growth that affects every long-running Lance writer, not just merge_insert users.

The fix: #7718

Cleanup (cleanup_old_versions, and optimize()'s auto-cleanup hook) already knows exactly which versions become unreachable when it deletes their files — it just never told the cache. #7718 wires that up: CleanupTask::run() now evicts the metadata/index cache entries for every version it physically deletes, gated on actual deletion (not explain()'s dry run).

The existing invalidate_prefix couldn't be reused as-is — it only matches an entry's namespace prefix (a dataset URI), never the per-entry key where version numbers live — so the PR adds a proper invalidate_key_prefix primitive with a /-boundary check (so evicting version 5 doesn't also evict 50) and includes a regression test that fails without the fix and passes with it.

Scoped to version-level entries for now, since that's what the evidence above points to as the dominant driver; per-fragment entries (DeletionFileKey, RowIdSequenceKey) are a smaller, documented follow-up. I'll open a separate PR for the original RAM-limit ask on merge_insert's join, since that's still worth doing independently — it just isn't sufficient on its own.

[yeung108]

Follow-up: opened #7719 for the original RAM-limit ask.

Turns out merge_insert has two join code paths, and only one had any memory bound at all — the indexed-scan path already ran through execute_plan() with use_spilling: true, but the full-table-scan path (create_full_table_joined_stream, taken whenever the join key isn't fully indexed, which includes the common no-index case from the original reproduction here) built its SessionContext with no memory limit or spilling whatsoever. #7719 fixes that and exposes MergeInsertBuilder.mem_pool_size() / .max_temp_directory_size() (Python and Rust) so it's configurable, per this issue's title.

To be clear about how the two PRs relate: #7718 fixes the dominant driver behind the RSS growth reported in this thread (a cache-invalidation gap that affects every writer, not just merge_insert, and doesn't go away with an index). #7719 fixes the narrower, original ask — bounding a single join's own transient memory. Both are worth having; neither alone is sufficient for the full picture.