diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md
index 3ac1dd031..4881d62b5 100644
--- a/docs/src/js/classes/Table.md
+++ b/docs/src/js/classes/Table.md
@@ -527,6 +527,8 @@ table's current checkout.
 
 Preserves input order, duplicates, and nulls. Use this for large payloads.
 See [Table.fetchBlobs](Table.md#fetchblobs) for row-ID validity across versions.
+One missing or deleted row ID rejects the entire batch, including on remote
+tables; no partial handles are returned.
 
 #### Parameters
 
@@ -556,6 +558,8 @@ Bytes for `column` at row IDs from [Query.withRowId](Query.md#withrowid).
 Reads the table's current checkout. IDs from another version can fail after
 compaction unless stable row ids are enabled. Results keep input order and
 duplicates. Null blobs are `null`. Empty blobs are empty buffers.
+One missing or deleted row ID rejects the entire batch, including on remote
+tables; no partial results are returned.
 Remote servers limit each request to 1024 row IDs and 64 MiB of blob bytes.
 The client splits requests automatically and reads an individual larger
 blob through the Range route. This method still materializes all bytes in
diff --git a/docs/src/python/python.md b/docs/src/python/python.md
index 3776283dc..28e5f9a1b 100644
--- a/docs/src/python/python.md
+++ b/docs/src/python/python.md
@@ -342,6 +342,8 @@ pylance, LanceDB uses a matching `lance.blob.v2` extension type so blob columns
 still work. Queries return descriptors. Call
 [`fetch_blob_files`][lancedb.table.Table.fetch_blob_files] for lazy reads or
 [`fetch_blobs`][lancedb.table.Table.fetch_blobs] for eager bytes.
+One missing or deleted row ID rejects the entire blob-read batch; the APIs do
+not return partial results.
 
 ::: lancedb.blob
 
diff --git a/nodejs/__test__/table.test.ts b/nodejs/__test__/table.test.ts
index d2244b48c..67d5a02d3 100644
--- a/nodejs/__test__/table.test.ts
+++ b/nodejs/__test__/table.test.ts
@@ -2817,6 +2817,18 @@ describe("when dealing with blob columns", () => {
     expect(Buffer.from(await files[1]!.read()).toString()).toBe("beta");
   });
 
+  it("rejects a whole blob read batch when one row ID was deleted", async () => {
+    const { table, rowIds } = await openBlobTable();
+    await table.delete("id = 2");
+
+    await expect(table.fetchBlobs("image", rowIds)).rejects.toThrow(
+      `first missing row ids: [${rowIds[1]}]`,
+    );
+    await expect(table.fetchBlobFiles("image", rowIds)).rejects.toThrow(
+      `first missing row ids: [${rowIds[1]}]`,
+    );
+  });
+
   it("adds a LargeBinary Arrow column to a blob table", async () => {
     const db = await connect(tmpDir.name);
     const schema = new Schema([new Field("id", new Int64()), blob("image")]);
diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts
index eae718168..dbf6d7ca2 100644
--- a/nodejs/lancedb/table.ts
+++ b/nodejs/lancedb/table.ts
@@ -536,6 +536,8 @@ export abstract class Table {
    * Reads the table's current checkout. IDs from another version can fail after
    * compaction unless stable row ids are enabled. Results keep input order and
    * duplicates. Null blobs are `null`. Empty blobs are empty buffers.
+   * One missing or deleted row ID rejects the entire batch, including on remote
+   * tables; no partial results are returned.
    * Remote servers limit each request to 1024 row IDs and 64 MiB of blob bytes.
    * The client splits requests automatically and reads an individual larger
    * blob through the Range route. This method still materializes all bytes in
@@ -553,6 +555,8 @@ export abstract class Table {
    *
    * Preserves input order, duplicates, and nulls. Use this for large payloads.
    * See {@link Table.fetchBlobs} for row-ID validity across versions.
+   * One missing or deleted row ID rejects the entire batch, including on remote
+   * tables; no partial handles are returned.
    */
   abstract fetchBlobFiles(
     column: string,
diff --git a/python/python/lancedb/_blob.py b/python/python/lancedb/_blob.py
index ee78a7312..0dba427cc 100644
--- a/python/python/lancedb/_blob.py
+++ b/python/python/lancedb/_blob.py
@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import io
+import operator
 from collections.abc import Awaitable, Callable, Iterable
 from typing import TYPE_CHECKING, Optional, Union
 
@@ -432,16 +433,30 @@ def _leaf_struct_column(tbl: pa.Table, path: str) -> pa.StructArray:
 
 
 def _normalize_blob_row_ids(
-    row_ids: Union[list[int], pa.Table], blob_column: str
+    row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table], blob_column: str
 ) -> list[int]:
     if isinstance(row_ids, pa.Table):
-        return read_row_ids_from_hits(row_ids, blob_column)
-    if isinstance(row_ids, (pa.Array, pa.ChunkedArray)):
-        raise ValueError(
-            "pass a query table with _rowid, not a column array "
-            "(use fetch_blobs('image', hits), not fetch_blobs('image', hits['image']))"
-        )
-    return list(row_ids)
+        values = read_row_ids_from_hits(row_ids, blob_column)
+    elif isinstance(row_ids, (pa.Array, pa.ChunkedArray)):
+        if not pa.types.is_integer(row_ids.type):
+            raise ValueError(
+                "row id arrays must have an integer type; pass a query table with "
+                "_rowid when using a blob descriptor column"
+            )
+        values = row_ids.to_pylist()
+    else:
+        values = list(row_ids)
+
+    normalized = []
+    for value in values:
+        try:
+            row_id = operator.index(value)
+        except TypeError as exc:
+            raise ValueError("row ids must be integers without nulls") from exc
+        if row_id < 0:
+            raise ValueError("row ids must be non-negative")
+        normalized.append(row_id)
+    return normalized
 
 
 def _wrap_blob_files(handles: Iterable[object]) -> list[Optional[BlobFile]]:
diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi
index 8d0a86e4a..4b0ebf839 100644
--- a/python/python/lancedb/_lancedb.pyi
+++ b/python/python/lancedb/_lancedb.pyi
@@ -1,6 +1,6 @@
 from datetime import date, datetime, timedelta
 from decimal import Decimal
-from typing import Dict, List, Optional, Tuple, Any, TypedDict, Union, Literal
+from typing import Dict, List, Optional, Sequence, Tuple, Any, TypedDict, Union, Literal
 from uuid import UUID
 
 import pyarrow as pa
@@ -563,7 +563,7 @@ class Table:
     async def fetch_blob_ranges(
         self,
         column: str,
-        requests: List[Tuple[int, int, int]],
+        requests: Sequence[Sequence[int]],
     ) -> pa.LargeBinaryArray: ...
     async def fetch_blob_files(
         self, column: str, row_ids: list[int]
diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py
index 7f03f8fa9..1bb238fbd 100644
--- a/python/python/lancedb/remote/table.py
+++ b/python/python/lancedb/remote/table.py
@@ -1154,7 +1154,9 @@ class RemoteTable(Table):
         return LOOP.run(self._table.blob_columns())
 
     def fetch_blobs(
-        self, column: str, row_ids: Union[list[int], pa.Table]
+        self,
+        column: str,
+        row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
     ) -> pa.LargeBinaryArray:
         return LOOP.run(self._table.fetch_blobs(column, row_ids))
 
@@ -1162,7 +1164,9 @@ class RemoteTable(Table):
         return LOOP.run(self._table.fetch_blob_ranges(column, list(requests)))
 
     def fetch_blob_files(
-        self, column: str, row_ids: Union[list[int], pa.Table]
+        self,
+        column: str,
+        row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
     ) -> "list[Optional[BlobFile]]":
         return LOOP.run(self._table.fetch_blob_files(column, row_ids))
 
diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py
index 81b376ebd..833d73c88 100644
--- a/python/python/lancedb/table.py
+++ b/python/python/lancedb/table.py
@@ -2058,15 +2058,20 @@ class Table(ABC):
 
     @abstractmethod
     def fetch_blobs(
-        self, column: str, row_ids: Union[list[int], pa.Table]
+        self,
+        column: str,
+        row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
     ) -> pa.LargeBinaryArray:
         """Materialize full blob bytes for ``column`` at the given rows.
 
         The result has the same length and order as ``row_ids``. Null blobs
         produce null slots; valid empty blobs produce ``b""``.
 
-        ``_rowid`` values stay valid after compaction when the table has stable
-        row ids.
+        ``row_ids`` may be a list, an integer Arrow array or chunked array,
+        or a query table with row ids. ``_rowid`` values stay valid after
+        compaction when the table has stable row ids.
+        One missing or deleted row ID rejects the entire batch; no partial
+        results are returned.
 
         Remote servers limit each request to 1024 row IDs and 64 MiB of blob
         bytes. The client splits requests automatically and reads an individual
@@ -2078,17 +2083,20 @@ class Table(ABC):
     def fetch_blob_ranges(
         self,
         column: str,
-        requests: Sequence[Tuple[int, int, int]],
+        requests: Sequence[Sequence[int]],
     ) -> pa.LargeBinaryArray:
         """Materialize row-specific byte ranges from a blob v2 column.
 
-        Each request is a ``(row_id, offset, length)`` tuple. Requests may be
-        repeated or reordered, including multiple ranges for the same blob.
+        Each request is a three-item ``(row_id, offset, length)`` sequence.
+        Requests may be repeated or reordered, including multiple ranges for
+        the same blob.
         The result has the same length and order as ``requests``; null blobs
         produce null slots and empty ranges on non-null blobs produce ``b""``.
 
         ``_rowid`` values stay valid after compaction when the table has stable
         row ids.
+        One missing or deleted row ID rejects the entire batch; no partial
+        results are returned.
 
         Row IDs can be obtained from a query with ``with_row_id(True)``. This
         API is currently supported only by local tables.
@@ -2096,18 +2104,22 @@ class Table(ABC):
 
     @abstractmethod
     def fetch_blob_files(
-        self, column: str, row_ids: Union[list[int], pa.Table]
+        self,
+        column: str,
+        row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
     ) -> "list[Optional[BlobFile]]":
         """Open lazy, seekable :class:`~lancedb._blob.BlobFile` handles.
 
         Prefer this over :meth:`fetch_blobs` for large payloads. ``row_ids`` is
-        a ``list[int]`` or a query ``pyarrow.Table`` carrying row identity via
-        ``_rowid`` or a ``_lance_row_id`` field on the blob descriptor. Null
-        rows are ``None``. Remote tables require LanceDB Cloud server 0.5.0 or
-        newer.
+        a ``list[int]``, an integer Arrow array or chunked array, or a query
+        ``pyarrow.Table`` carrying row identity via ``_rowid`` or a
+        ``_lance_row_id`` field on the blob descriptor. Null blob values return
+        ``None``. Remote tables require LanceDB Cloud server 0.5.0 or newer.
 
         ``_rowid`` values stay valid after compaction when the table has stable
         row ids.
+        One missing or deleted row ID rejects the entire batch; no partial
+        handles are returned.
         """
 
     @abstractmethod
@@ -3013,19 +3025,23 @@ class LanceTable(Table):
         return LOOP.run(self._table.blob_columns())
 
     def fetch_blobs(
-        self, column: str, row_ids: Union[list[int], pa.Table]
+        self,
+        column: str,
+        row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
     ) -> pa.LargeBinaryArray:
         return LOOP.run(self._table.fetch_blobs(column, row_ids))
 
     def fetch_blob_ranges(
         self,
         column: str,
-        requests: Sequence[Tuple[int, int, int]],
+        requests: Sequence[Sequence[int]],
     ) -> pa.LargeBinaryArray:
         return LOOP.run(self._table.fetch_blob_ranges(column, list(requests)))
 
     def fetch_blob_files(
-        self, column: str, row_ids: Union[list[int], pa.Table]
+        self,
+        column: str,
+        row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
     ) -> "list[Optional[BlobFile]]":
         return LOOP.run(self._table.fetch_blob_files(column, row_ids))
 
@@ -7079,7 +7095,9 @@ class AsyncTable:
         return await self._inner.blob_columns()
 
     async def fetch_blobs(
-        self, column: str, row_ids: Union[list[int], pa.Table]
+        self,
+        column: str,
+        row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
     ) -> pa.LargeBinaryArray:
         return await self._inner.fetch_blobs(
             column, _normalize_blob_row_ids(row_ids, column)
@@ -7088,12 +7106,14 @@ class AsyncTable:
     async def fetch_blob_ranges(
         self,
         column: str,
-        requests: Sequence[Tuple[int, int, int]],
+        requests: Sequence[Sequence[int]],
     ) -> pa.LargeBinaryArray:
         return await self._inner.fetch_blob_ranges(column, list(requests))
 
     async def fetch_blob_files(
-        self, column: str, row_ids: Union[list[int], pa.Table]
+        self,
+        column: str,
+        row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
     ) -> "list[Optional[BlobFile]]":
         handles = await self._inner.fetch_blob_files(
             column, _normalize_blob_row_ids(row_ids, column)
diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py
index 27e920e50..f852f28f5 100644
--- a/python/python/tests/test_blob.py
+++ b/python/python/tests/test_blob.py
@@ -70,21 +70,25 @@ def _row_ids_by_id(table):
 
 def _assert_missing_blob_row_ids(exc_info):
     message = str(exc_info.value)
-    assert "row ids" in message
+    assert "row ids but some do not exist in the table" in message
     assert "rowaddr" not in message
     assert "fragment" not in message
 
 
 def _assert_fetch_apis_reject_missing_row_ids(table, row_ids):
+    expected = f"first missing row ids: [{row_ids[0]}]"
     with pytest.raises(ValueError) as exc_info:
         table.fetch_blobs("image", row_ids)
     _assert_missing_blob_row_ids(exc_info)
+    assert expected in str(exc_info.value)
     with pytest.raises(ValueError) as exc_info:
         table.fetch_blob_files("image", row_ids)
     _assert_missing_blob_row_ids(exc_info)
+    assert expected in str(exc_info.value)
     with pytest.raises(ValueError) as exc_info:
         table.fetch_blob_ranges("image", [(row_id, 0, 1) for row_id in row_ids])
     _assert_missing_blob_row_ids(exc_info)
+    assert expected in str(exc_info.value)
 
 
 def test_blob_factory_declares_v2_field():
@@ -861,6 +865,38 @@ def test_fetch_blobs_accepts_query_result():
     assert {blobs[i].as_py() for i in range(len(blobs))} == {b"gamma"}
 
 
+def test_fetch_blob_apis_accept_integer_arrow_row_ids():
+    table = _blob_table(
+        "arrow_row_ids",
+        [{"id": 1, "image": b"alpha"}, {"id": 2, "image": b"beta"}],
+    )
+    by_id = _row_ids_by_id(table)
+    ids = [by_id[2], by_id[1]]
+
+    for row_ids in (
+        pa.array(ids, type=pa.uint64()),
+        pa.chunked_array([ids[:1], ids[1:]], type=pa.int64()),
+    ):
+        assert table.fetch_blobs("image", row_ids).to_pylist() == [b"beta", b"alpha"]
+        files = table.fetch_blob_files("image", row_ids)
+        assert [file.read() for file in files] == [b"beta", b"alpha"]
+
+
+def test_fetch_blob_apis_validate_arrow_and_negative_row_ids():
+    table = _blob_table("invalid_arrow_row_ids", [{"id": 1, "image": b"alpha"}])
+    hits = table.search().with_row_id(True).to_arrow()
+
+    for fetch in (table.fetch_blobs, table.fetch_blob_files):
+        with pytest.raises(ValueError, match="non-negative"):
+            fetch("image", [-1])
+        with pytest.raises(ValueError, match="non-negative"):
+            fetch("image", pa.array([-1], type=pa.int64()))
+        with pytest.raises(ValueError, match="without nulls"):
+            fetch("image", pa.array([None], type=pa.int64()))
+        with pytest.raises(ValueError, match="integer type"):
+            fetch("image", hits["image"])
+
+
 def test_fetch_blobs_after_compact_with_stable_row_ids(tmp_path):
     db = lancedb.connect(
         tmp_path, storage_options={"new_table_enable_stable_row_ids": "true"}
@@ -992,6 +1028,15 @@ def test_fetch_blob_ranges_aligns_repeated_ranges_and_nulls():
     assert ranges.to_pylist() == [b"cde", None, b"ab", b"cde", b""]
 
 
+def test_fetch_blob_ranges_accepts_three_item_sequences():
+    table = _blob_table("range_sequences", [{"id": 1, "image": b"abcdef"}])
+    row_id = _row_ids_by_id(table)[1]
+
+    ranges = table.fetch_blob_ranges("image", [[row_id, 1, 2], (row_id, 4, 2)])
+
+    assert ranges.to_pylist() == [b"bc", b"ef"]
+
+
 def test_fetch_blob_ranges_validates_requests():
     table = _blob_table("range_validation", [{"id": 1, "image": b"abc"}])
     row_id = _row_ids_by_id(table)[1]
@@ -1006,6 +1051,30 @@ def test_fetch_blob_ranges_validates_requests():
         table.fetch_blob_ranges("image", [(2**64 - 1, 0, 1)])
     _assert_missing_blob_row_ids(exc_info)
 
+    for index, name in enumerate(("row id", "offset", "length")):
+        request = [row_id, 0, 1]
+        request[index] = -1
+        with pytest.raises(
+            ValueError, match=f"blob range {name} must be a non-negative"
+        ):
+            table.fetch_blob_ranges("image", [request])
+
+    with pytest.raises(ValueError, match="three values"):
+        table.fetch_blob_ranges("image", [[row_id, 0]])
+
+
+def test_fetch_blob_apis_reject_row_past_fragment_end():
+    table = _blob_table(
+        "past_fragment_end",
+        [
+            {"id": 1, "image": b"a"},
+            {"id": 2, "image": b"b"},
+            {"id": 3, "image": b"c"},
+        ],
+    )
+    _assert_fetch_apis_reject_missing_row_ids(table, [3])
+    _assert_fetch_apis_reject_missing_row_ids(table, [3, 0])
+
 
 def test_fetch_blob_apis_reject_missing_fragment_row_addr():
     table = _blob_table("missing_frag", [{"id": 1, "image": b"x"}])
@@ -1023,6 +1092,19 @@ def test_fetch_blob_apis_reject_deleted_row_ids():
     _assert_fetch_apis_reject_missing_row_ids(table, [by_id[2], by_id[1]])
 
 
+def test_fetch_blob_apis_reject_deleted_stable_row_ids(tmp_path):
+    db = lancedb.connect(
+        tmp_path, storage_options={"new_table_enable_stable_row_ids": "true"}
+    )
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    table = db.create_table("stable_deleted", schema=schema)
+    table.add([{"id": 1, "image": b"one"}, {"id": 2, "image": b"two"}])
+    by_id = _row_ids_by_id(table)
+    table.delete("id = 2")
+
+    _assert_fetch_apis_reject_missing_row_ids(table, [by_id[2], by_id[1]])
+
+
 def test_fetch_blob_ranges_empty_requests_returns_empty_array():
     table = _blob_table("range_empty", [{"id": 1, "image": b"x"}])
     assert table.fetch_blob_ranges("image", []).to_pylist() == []
diff --git a/python/src/table.rs b/python/src/table.rs
index d82ebbaaf..7c0ec6610 100644
--- a/python/src/table.rs
+++ b/python/src/table.rs
@@ -1462,14 +1462,30 @@ impl Table {
     pub fn fetch_blob_ranges(
         self_: PyRef<'_, Self>,
         column: String,
-        requests: Vec<(u64, u64, u64)>,
+        requests: Vec<Vec<i128>>,
     ) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner_ref()?.clone();
+        let requests = requests
+            .into_iter()
+            .map(|values| {
+                let [row_id, offset, length] = <[i128; 3]>::try_from(values).map_err(|_| {
+                    PyValueError::new_err("each blob range request must have three values")
+                })?;
+                let to_u64 = |value, name| {
+                    u64::try_from(value).map_err(|_| {
+                        PyValueError::new_err(format!(
+                            "blob range {name} must be a non-negative 64-bit integer"
+                        ))
+                    })
+                };
+                Ok(BlobRangeRequest::new(
+                    to_u64(row_id, "row id")?,
+                    to_u64(offset, "offset")?,
+                    to_u64(length, "length")?,
+                ))
+            })
+            .collect::<PyResult<Vec<_>>>()?;
         future_into_py(self_.py(), async move {
-            let requests = requests
-                .into_iter()
-                .map(|(row_id, offset, length)| BlobRangeRequest::new(row_id, offset, length))
-                .collect::<Vec<_>>();
             let blobs: LargeBinaryArray = inner
                 .fetch_blob_ranges(column, requests)
                 .await
diff --git a/rust/lancedb/src/blob.rs b/rust/lancedb/src/blob.rs
index b7f16974e..178752952 100644
--- a/rust/lancedb/src/blob.rs
+++ b/rust/lancedb/src/blob.rs
@@ -399,45 +399,115 @@ pub(crate) fn ensure_blob_v2_column(
     }
 }
 
-fn ensure_all_row_ids_resolved(column: &str, requested: usize, resolved: usize) -> Result<()> {
-    if requested == resolved {
+fn missing_blob_row_ids_error(column: &str, requested: usize, missing: &[u64]) -> Error {
+    let detail = if missing.is_empty() {
+        String::new()
+    } else {
+        format!("; first missing row ids: {missing:?}")
+    };
+    Error::InvalidInput {
+        message: format!(
+            "blob read for column '{column}' requested {requested} row ids but some \
+             do not exist in the table{detail}; pass row ids collected from this table"
+        ),
+    }
+}
+
+/// Diagnose only after a failed read. Stable row IDs use Lance's deletion-aware
+/// index; physical row IDs are fragment/offset pairs. Limit the error to three
+/// distinct IDs so a large failed batch stays readable.
+async fn find_missing_blob_row_ids(dataset: &Dataset, row_ids: &[u64]) -> Result<Vec<u64>> {
+    let mut missing = Vec::new();
+    if let Some(index) = lance::dataset::rowids::get_row_id_index(dataset).await? {
+        for (row_id, address) in row_ids.iter().zip(index.get_many(row_ids)?) {
+            if address.is_none() && !missing.contains(row_id) {
+                missing.push(*row_id);
+                if missing.len() == 3 {
+                    break;
+                }
+            }
+        }
+    } else {
+        let mut fragments = std::collections::HashMap::new();
+        for &row_id in row_ids {
+            let fragment_id = row_id >> 32;
+            let offset = row_id as u32;
+            if let std::collections::hash_map::Entry::Vacant(entry) = fragments.entry(fragment_id) {
+                let fragment = dataset.get_fragment(fragment_id as usize);
+                let state = if let Some(fragment) = fragment {
+                    Some((
+                        fragment.physical_rows().await?,
+                        fragment.get_deletion_vector().await?,
+                    ))
+                } else {
+                    None
+                };
+                entry.insert(state);
+            }
+            let is_missing = match fragments.get(&fragment_id) {
+                Some(Some((physical_rows, deletions))) => {
+                    offset as usize >= *physical_rows
+                        || deletions
+                            .as_ref()
+                            .is_some_and(|deletions| deletions.contains(offset))
+                }
+                Some(None) => true,
+                None => unreachable!(),
+            };
+            if is_missing && !missing.contains(&row_id) {
+                missing.push(row_id);
+                if missing.len() == 3 {
+                    break;
+                }
+            }
+        }
+    }
+    Ok(missing)
+}
+
+async fn ensure_all_row_ids_resolved(
+    dataset: &Dataset,
+    column: &str,
+    row_ids: &[u64],
+    resolved: usize,
+) -> Result<()> {
+    if row_ids.len() == resolved {
         return Ok(());
     }
-    if resolved < requested {
-        Err(Error::InvalidInput {
-            message: format!(
-                "blob read for column '{column}' requested {requested} row ids but only {resolved} \
-                 exist in the table; pass row ids collected from this table"
-            ),
-        })
+    if resolved < row_ids.len() {
+        let missing = find_missing_blob_row_ids(dataset, row_ids)
+            .await
+            .unwrap_or_default();
+        Err(missing_blob_row_ids_error(column, row_ids.len(), &missing))
     } else {
         Err(Error::Runtime {
             message: format!(
-                "blob read for column '{column}' returned {resolved} results for {requested} row ids"
+                "blob read for column '{column}' returned {resolved} results for {} row ids",
+                row_ids.len()
             ),
         })
     }
 }
 
-/// Lance take reports a missing physical row address as NotSupported or InvalidInput.
-fn map_blob_take_error(column: &str, requested: usize, err: lance::Error) -> Error {
-    let missing_row_addr = match &err {
-        lance::Error::NotSupported { source, .. } => {
-            source.to_string().contains("must not target deleted rows")
-        }
-        lance::Error::InvalidInput { source, .. } => source
-            .to_string()
-            .contains("belongs to non-existent fragment"),
-        _ => false,
-    };
+/// Lance reports missing physical and stable row ids through several error paths
+/// without a shared typed error.
+async fn map_blob_take_error(
+    dataset: &Dataset,
+    column: &str,
+    row_ids: &[u64],
+    err: lance::Error,
+) -> Error {
+    let message = err.to_string();
+    let missing_row_id = message.contains("must not target deleted rows")
+        || message.contains("belongs to non-existent fragment")
+        || (message.contains("Invalid read params ") && message.contains("addressable rows"))
+        || message.contains("Could not resolve all requested row IDs");
 
-    if missing_row_addr {
-        Error::InvalidInput {
-            message: format!(
-                "blob read for column '{column}' requested {requested} row ids but some \
-                 do not exist in the table; pass row ids collected from this table"
-            ),
-        }
+    if missing_row_id {
+        let missing = find_missing_blob_row_ids(dataset, row_ids)
+            .await
+            .unwrap_or_default();
+        missing_blob_row_ids_error(column, row_ids.len(), &missing)
     } else {
         err.into()
     }
@@ -454,18 +524,25 @@ pub(crate) async fn take_blob_ranges_aligned(
         return Ok(LargeBinaryBuilder::new().finish());
     }
 
+    let row_ids = requests
+        .iter()
+        .map(|request| request.row_id)
+        .collect::<Vec<_>>();
     let lance_requests = requests
         .iter()
         .map(|request| LanceBlobRangeRequest::new(request.row_id, request.offset, request.length))
         .collect::<Vec<_>>();
-    let payloads = dataset
+    let payloads = match dataset
         .read_blob_ranges(column)?
         .with_row_ids(lance_requests)
         .preserve_order(true)
         .execute()
         .await
-        .map_err(|err| map_blob_take_error(column, requests.len(), err))?;
-    ensure_all_row_ids_resolved(column, requests.len(), payloads.len())?;
+    {
+        Ok(payloads) => payloads,
+        Err(err) => return Err(map_blob_take_error(dataset, column, &row_ids, err).await),
+    };
+    ensure_all_row_ids_resolved(dataset, column, &row_ids, payloads.len()).await?;
 
     let mut builder = LargeBinaryBuilder::new();
     for payload in payloads {
@@ -488,14 +565,17 @@ pub(crate) async fn take_blobs_aligned(
         return Ok(LargeBinaryBuilder::new().finish());
     }
 
-    let payloads = dataset
+    let payloads = match dataset
         .read_blobs(column)?
         .with_row_ids(row_ids.to_vec())
         .preserve_order(true)
         .execute()
         .await
-        .map_err(|err| map_blob_take_error(column, row_ids.len(), err))?;
-    ensure_all_row_ids_resolved(column, row_ids.len(), payloads.len())?;
+    {
+        Ok(payloads) => payloads,
+        Err(err) => return Err(map_blob_take_error(dataset, column, row_ids, err).await),
+    };
+    ensure_all_row_ids_resolved(dataset, column, row_ids, payloads.len()).await?;
 
     let mut builder = LargeBinaryBuilder::new();
     for payload in payloads {
@@ -518,11 +598,11 @@ pub(crate) async fn take_blob_files_aligned(
         return Ok(Vec::new());
     }
 
-    let handles = dataset
-        .take_blobs(row_ids, column)
-        .await
-        .map_err(|err| map_blob_take_error(column, row_ids.len(), err))?;
-    ensure_all_row_ids_resolved(column, row_ids.len(), handles.len())?;
+    let handles = match dataset.take_blobs(row_ids, column).await {
+        Ok(handles) => handles,
+        Err(err) => return Err(map_blob_take_error(dataset, column, row_ids, err).await),
+    };
+    ensure_all_row_ids_resolved(dataset, column, row_ids, handles.len()).await?;
     Ok(handles
         .into_iter()
         .map(|handle| handle.map(Into::into))
diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs
index 15ca3702b..3a833b293 100644
--- a/rust/lancedb/src/table.rs
+++ b/rust/lancedb/src/table.rs
@@ -1259,7 +1259,8 @@ impl Table {
     /// [`Self::fetch_blob_files`] for large selections.
     ///
     /// `_rowid` values stay valid after compaction when the table has stable
-    /// row ids.
+    /// row ids. One missing or deleted row id rejects the whole request; no
+    /// partial results are returned.
     ///
     /// ```
     /// use arrow_array::UInt64Array;
@@ -1303,7 +1304,8 @@ impl Table {
     /// non-null blobs produce empty byte strings.
     ///
     /// `_rowid` values stay valid after compaction when the table has stable
-    /// row ids.
+    /// row ids. One missing or deleted row id rejects the whole request; no
+    /// partial results are returned.
     ///
     /// ```
     /// use lancedb::blob::BlobRangeRequest;
@@ -1344,7 +1346,8 @@ impl Table {
     /// read from disk until a call to [`BlobFile::read`].
     ///
     /// `_rowid` values stay valid after compaction when the table has stable
-    /// row ids.
+    /// row ids. One missing or deleted row id rejects the whole request; no
+    /// partial handles are returned.
     ///
     /// ```
     /// # use lancedb::Table;
diff --git a/rust/lancedb/tests/blob_integration.rs b/rust/lancedb/tests/blob_integration.rs
index dd4051657..0c9942db6 100644
--- a/rust/lancedb/tests/blob_integration.rs
+++ b/rust/lancedb/tests/blob_integration.rs
@@ -501,17 +501,23 @@ async fn collect_id_rowid(table: &Table) -> Result<Vec<(i64, u64)>> {
 fn assert_missing_blob_row_ids(err: &Error) {
     assert!(matches!(err, Error::InvalidInput { .. }), "got {err:?}");
     let message = err.to_string();
-    assert!(message.contains("row ids"), "{message}");
+    assert!(
+        message.contains("row ids but some do not exist in the table"),
+        "{message}"
+    );
     assert!(!message.contains("rowaddr"), "{message}");
     assert!(!message.contains("fragment"), "{message}");
 }
 
 async fn assert_fetch_apis_reject_missing_row_ids(table: &Table, row_ids: &[u64]) -> Result<()> {
+    let expected = format!("first missing row ids: [{}]", row_ids[0]);
     let err = table.fetch_blobs("image", row_ids).await.unwrap_err();
     assert_missing_blob_row_ids(&err);
+    assert!(err.to_string().contains(&expected), "{err}");
 
     let err = table.fetch_blob_files("image", row_ids).await.unwrap_err();
     assert_missing_blob_row_ids(&err);
+    assert!(err.to_string().contains(&expected), "{err}");
 
     let err = table
         .fetch_blob_ranges(
@@ -524,6 +530,7 @@ async fn assert_fetch_apis_reject_missing_row_ids(table: &Table, row_ids: &[u64]
         .await
         .unwrap_err();
     assert_missing_blob_row_ids(&err);
+    assert!(err.to_string().contains(&expected), "{err}");
     Ok(())
 }
 
@@ -790,6 +797,30 @@ async fn fetch_blobs_out_of_range_id_errors_without_panic() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn fetch_blob_apis_reject_row_past_fragment_end() -> Result<()> {
+    let tmp = tempdir().unwrap();
+    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
+    let table =
+        create_inline_blob_table(&db, "t", &[1, 2, 3], &[Some(b"a"), Some(b"b"), Some(b"c")])
+            .await?;
+
+    assert_fetch_apis_reject_missing_row_ids(&table, &[3]).await?;
+    assert_fetch_apis_reject_missing_row_ids(&table, &[3, 0]).await
+}
+
+#[tokio::test]
+async fn fetch_blobs_identifies_multiple_missing_ids_without_partial_results() -> Result<()> {
+    let tmp = tempdir().unwrap();
+    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
+    let table = create_inline_blob_table(&db, "t", &[1, 2], &[Some(b"a"), Some(b"b")]).await?;
+
+    let err = table.fetch_blobs("image", &[0, 2, 3]).await.unwrap_err();
+    assert_missing_blob_row_ids(&err);
+    assert!(err.to_string().contains("first missing row ids: [2, 3]"));
+    Ok(())
+}
+
 #[tokio::test]
 async fn fetch_blob_files_rejects_missing_fragment_row_addr() -> Result<()> {
     let tmp = tempdir().unwrap();
@@ -831,6 +862,23 @@ async fn fetch_blob_apis_reject_deleted_row_ids() -> Result<()> {
     assert_fetch_apis_reject_missing_row_ids(&table, &[deleted_row_addr, live_row_addr]).await
 }
 
+#[tokio::test]
+async fn fetch_blob_apis_reject_deleted_stable_row_ids() -> Result<()> {
+    let tmp = tempdir().unwrap();
+    let db = connect(tmp.path().to_str().unwrap())
+        .storage_option(OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, "true")
+        .execute()
+        .await?;
+    let table = create_inline_blob_table(&db, "t", &[1, 2], &[Some(b"one"), Some(b"two")]).await?;
+    let pairs = collect_id_rowid(&table).await?;
+    let deleted_row_id = pairs.iter().find(|(id, _)| *id == 2).unwrap().1;
+    let live_row_id = pairs.iter().find(|(id, _)| *id == 1).unwrap().1;
+
+    table.delete("id = 2").await?;
+
+    assert_fetch_apis_reject_missing_row_ids(&table, &[deleted_row_id, live_row_id]).await
+}
+
 #[tokio::test]
 async fn fetch_blobs_rejects_non_blob_column() -> Result<()> {
     let tmp = tempdir().unwrap();