fix(python): reject unsafe remote blob materialization

This commit is contained in:
Gatefixer
2026-09-23 19:39:55 +00:00
parent b1d9c4cb84
commit b9dbaef0ad
3 changed files with 64 additions and 68 deletions
+9 -49
View File
@@ -282,24 +282,6 @@ def _scanner_to_pandas(
return tbl.to_pandas(**kwargs)
def _set_blob_frame_values(df: "pd.DataFrame", path: str, values: list) -> None:
"""Replace a blob output column, including a blob inside a struct column."""
if path in df.columns:
df[path] = values
return
top, *nested = path.split(".")
if top not in df.columns or not nested:
raise ValueError(f"blob output column {path!r} is missing from query results")
for record, value in zip(df[top], values):
for part in nested[:-1]:
if record is None:
break
record = record[part]
if record is not None:
record[nested[-1]] = value
def _finish_plain_scan_pandas(
scanner: _LanceScanner,
*,
@@ -3038,8 +3020,9 @@ class AsyncQueryBase(object):
complete within the specified time, an error will be raised.
blob_mode: str, default "lazy"
Controls how blob columns are returned. Local plain scan queries
use Lance native conversion; remote queries use server blob fetch
APIs for blob v2 "bytes" and "lazy" modes.
use Lance native conversion. Remote queries support descriptions,
but cannot safely materialize bytes or lazy handles without a
stable table snapshot across the query and blob fetches.
**kwargs
Forwarded to pyarrow.Table.to_pandas after query execution and
optional flattening.
@@ -3085,10 +3068,8 @@ class AsyncQueryBase(object):
blob_mode: BlobMode,
**kwargs,
) -> "pd.DataFrame":
# to_arrow() runs the query on the server and keeps the row ids needed
# by the server-side blob fetch APIs in the v2 descriptors.
table = self._table
assert table is not None
# A live remote table can be replaced between query and blob fetch.
# Row ids and version numbers do not identify the producing table.
tbl = await self.to_arrow(timeout=timeout)
projected_blob_paths = set(blob_column_paths(tbl.schema))
if not projected_blob_paths:
@@ -3098,32 +3079,11 @@ class AsyncQueryBase(object):
tbl = strip_auto_row_ids(tbl, self._blob_paths)
return flatten_columns(tbl, flatten).to_pandas(**kwargs)
schema = await table.schema()
blob_sources = blob_v2_projection_sources(
schema, self.to_query_object().columns
raise NotImplementedError(
f"remote to_pandas(blob_mode={blob_mode!r}) cannot safely materialize "
"blob columns without a stable table snapshot; "
"use blob_mode='descriptions'"
)
if not projected_blob_paths.issubset(blob_sources):
raise NotImplementedError(
f"remote to_pandas(blob_mode={blob_mode!r}) requires row-addressable "
"blob v2 columns"
)
values = {}
for output_name, source_name in blob_sources.items():
if output_name not in projected_blob_paths:
continue
if blob_mode == "bytes":
values[output_name] = (
await table.fetch_blobs(source_name, tbl)
).to_pylist()
else:
values[output_name] = await table.fetch_blob_files(source_name, tbl)
tbl = strip_auto_row_ids(tbl, self._blob_paths)
df = flatten_columns(tbl, flatten).to_pandas(**kwargs)
for output_name, column_values in values.items():
_set_blob_frame_values(df, output_name, column_values)
return df
async def _plain_scan_to_pandas(
self,
+3 -2
View File
@@ -5479,8 +5479,9 @@ class AsyncTable:
Parameters
----------
blob_mode: str, default "lazy"
Controls how Lance blob columns are returned. On remote tables,
blob v2 data is fetched through the server for "bytes" and "lazy".
Controls how Lance blob columns are returned. Remote tables support
"descriptions"; "bytes" and "lazy" require a stable table snapshot
that the server does not yet provide.
**kwargs
Forwarded to PyArrow / Lance pandas conversion.
+52 -17
View File
@@ -2421,9 +2421,13 @@ def blob_remote_table(*, server_version=Version("0.5.0")):
elif request.path == "/v1/table/test/query/":
content_len = int(request.headers.get("Content-Length", 0))
body = json.loads(request.rfile.read(content_len))
assert body.get("columns") in (None, ["id", "image"])
assert body["with_row_id"] is True
response_table = blob_query_response_table()
columns = body.get("columns")
assert columns in (None, ["id", "image"], ["id"])
if columns == ["id"]:
response_table = blob_query_response_table().select(["id"])
else:
assert body["with_row_id"] is True
response_table = blob_query_response_table()
request.send_response(200)
request.send_header("Content-Type", "application/vnd.apache.arrow.file")
request.end_headers()
@@ -2471,8 +2475,7 @@ async def test_async_remote_to_lance_rejects_storage_access(monkeypatch):
@pytest.mark.asyncio
@pytest.mark.parametrize("blob_mode", ["descriptions", "bytes", "lazy"])
async def test_async_remote_blob_to_pandas_uses_server(monkeypatch, blob_mode):
async def test_async_remote_blob_descriptions_to_pandas_uses_server(monkeypatch):
with blob_remote_table() as remote_table:
table = remote_table._table
@@ -2485,25 +2488,57 @@ async def test_async_remote_blob_to_pandas_uses_server(monkeypatch, blob_mode):
monkeypatch.setattr(table, "uri", fail_uri)
monkeypatch.setattr(table, "to_lance", fail_to_lance)
pandas_kwargs = {} if blob_mode == "lazy" else {"blob_mode": blob_mode}
query_df = (
await table.query().select(["id", "image"]).to_pandas(**pandas_kwargs)
query_df = await (
table.query().select(["id", "image"]).to_pandas(blob_mode="descriptions")
)
table_df = await table.to_pandas(**pandas_kwargs)
table_df = await table.to_pandas(blob_mode="descriptions")
for df in (query_df, table_df):
assert df["id"].tolist() == [1, 2, 3]
assert "_rowid" not in df.columns
images = df["image"].tolist()
assert images[1] is None
if blob_mode == "bytes":
assert images == [b"alpha", None, b"gamma"]
elif blob_mode == "lazy":
assert await images[0].aread() == b"alpha"
assert await images[2].aread() == b"gamma"
else:
assert images[0]["size"] == 5
assert "_lance_row_id" not in images[0]
assert images[0]["size"] == 5
assert "_lance_row_id" not in images[0]
@pytest.mark.asyncio
@pytest.mark.parametrize("blob_mode", ["bytes", "lazy"])
async def test_async_remote_blob_pandas_rejects_unpinned_fetch(monkeypatch, blob_mode):
with blob_remote_table() as remote_table:
table = remote_table._table
async def fail_uri():
raise AssertionError("remote table URI must not be requested")
async def fail_to_lance():
raise AssertionError("remote table storage must not be opened")
async def fail_fetch(*args, **kwargs):
raise AssertionError("blob fetch must not use an unpinned remote table")
monkeypatch.setattr(table, "uri", fail_uri)
monkeypatch.setattr(table, "to_lance", fail_to_lance)
monkeypatch.setattr(table, "fetch_blobs", fail_fetch)
monkeypatch.setattr(table, "fetch_blob_files", fail_fetch)
pandas_kwargs = {} if blob_mode == "lazy" else {"blob_mode": blob_mode}
with pytest.raises(NotImplementedError, match="stable table snapshot"):
await table.query().select(["id", "image"]).to_pandas(**pandas_kwargs)
with pytest.raises(NotImplementedError, match="stable table snapshot"):
await table.to_pandas(**pandas_kwargs)
@pytest.mark.asyncio
@pytest.mark.parametrize("blob_mode", ["bytes", "lazy"])
async def test_async_remote_nonblob_projection_to_pandas(blob_mode):
with blob_remote_table() as remote_table:
df = await (
remote_table._table.query().select(["id"]).to_pandas(blob_mode=blob_mode)
)
assert df["id"].tolist() == [1, 2, 3]
assert list(df.columns) == ["id"]
def test_remote_blob_files_are_lazy_seekable_handles():