mirror of
https://github.com/lancedb/lancedb.git
synced 2026-08-18 20:18:37 +00:00
test(python): cover Arrow buffer release after add (#3860)
## Summary - add deterministic regression coverage that `Table.add()` releases backing Arrow buffers without cyclic garbage collection - track the foreign buffer owner rather than RSS, separating live input retention from allocator high-water behavior - preserve the bounded-lifetime behavior of the Scannable writer that superseded the historical preprocessing path ## Root cause The historical Python preprocessing/write path produced a high allocator RSS while ingesting very wide IPC batches. The current Scannable writer releases each input buffer when `Table.add()` completes; remaining RSS is allocator high-water rather than a live Arrow reference. The resolved behavior had no regression coverage, so a future native lifetime regression could silently reintroduce the original failure mode. ## Validation - `uv run --extra tests --extra dev maturin develop` - `uv run --project python --extra tests pytest python/python/tests/test_table.py::test_add python/python/tests/test_table.py::test_add_releases_arrow_buffers_without_gc -q` - `uv run --project python --extra dev ruff format --check python/python/tests/test_table.py` - `uv run --project python --extra dev ruff check .` Fixes #2512 <!-- lance-gatekeeper-fix:v1 agent=29226408a8d07da592daf341d5384e37 generation=1 --> Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
This commit is contained in:
committed by
GitHub
parent
ac8b28c010
commit
2c06a48bd8
@@ -2,10 +2,13 @@
|
||||
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
|
||||
import ctypes
|
||||
import gc
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import warnings
|
||||
import weakref
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import date, datetime, timedelta
|
||||
from time import sleep
|
||||
@@ -459,6 +462,38 @@ def test_add(mem_db: DBConnection):
|
||||
_add(table, schema)
|
||||
|
||||
|
||||
def test_add_releases_arrow_buffers_without_gc(mem_db: DBConnection):
|
||||
"""Regression test for https://github.com/lancedb/lancedb/issues/2512."""
|
||||
schema = pa.schema([pa.field("x", pa.int64())])
|
||||
table = mem_db.create_table("test_add_releases_arrow_buffers", schema=schema)
|
||||
|
||||
class BufferOwner:
|
||||
def __init__(self, size: int):
|
||||
self.memory = ctypes.create_string_buffer(size)
|
||||
|
||||
owner_refs = []
|
||||
gc_was_enabled = gc.isenabled()
|
||||
gc.disable()
|
||||
try:
|
||||
for _ in range(3):
|
||||
size = 8 * 1024
|
||||
owner = BufferOwner(size)
|
||||
arrow_buffer = pa.foreign_buffer(
|
||||
ctypes.addressof(owner.memory), size, owner
|
||||
)
|
||||
array = pa.Array.from_buffers(pa.int64(), 1024, [None, arrow_buffer])
|
||||
batch = pa.RecordBatch.from_arrays([array], schema=schema)
|
||||
owner_refs.append(weakref.ref(owner))
|
||||
|
||||
table.add(batch)
|
||||
del batch, array, arrow_buffer, owner
|
||||
|
||||
assert all(owner_ref() is None for owner_ref in owner_refs)
|
||||
finally:
|
||||
if gc_was_enabled:
|
||||
gc.enable()
|
||||
|
||||
|
||||
def test_add_write_parallelism(mem_db: DBConnection):
|
||||
schema = pa.schema([pa.field("id", pa.int64())])
|
||||
table = mem_db.create_table("test", schema=schema)
|
||||
|
||||
Reference in New Issue
Block a user