diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py
index e397272fc..127ad3722 100644
--- a/python/python/lancedb/table.py
+++ b/python/python/lancedb/table.py
@@ -1793,6 +1793,9 @@ class Table(ABC):
         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.
+
         Convenience for small payloads. For large values use
         :meth:`fetch_blob_files`.
         """
@@ -1810,6 +1813,9 @@ class Table(ABC):
         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.
+
         Row IDs can be obtained from a query with ``with_row_id(True)``. This
         API is currently supported only by local tables.
         """
@@ -1825,6 +1831,9 @@ class Table(ABC):
         ``_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.
+
+        ``_rowid`` values stay valid after compaction when the table has stable
+        row ids.
         """
 
     @abstractmethod
diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py
index c9694277c..1f158fb49 100644
--- a/python/python/tests/test_blob.py
+++ b/python/python/tests/test_blob.py
@@ -66,6 +66,25 @@ def _row_ids_by_id(table):
     return dict(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))
 
 
+def _assert_missing_blob_row_ids(exc_info):
+    message = str(exc_info.value)
+    assert "row ids" in message
+    assert "rowaddr" not in message
+    assert "fragment" not in message
+
+
+def _assert_fetch_apis_reject_missing_row_ids(table, row_ids):
+    with pytest.raises(ValueError) as exc_info:
+        table.fetch_blobs("image", row_ids)
+    _assert_missing_blob_row_ids(exc_info)
+    with pytest.raises(ValueError) as exc_info:
+        table.fetch_blob_files("image", row_ids)
+    _assert_missing_blob_row_ids(exc_info)
+    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)
+
+
 def test_blob_factory_declares_v2_field():
     field = lancedb.blob("image")
     assert isinstance(field.type, pa.ExtensionType)
@@ -691,6 +710,25 @@ def test_fetch_blobs_accepts_query_result():
     assert {blobs[i].as_py() for i in range(len(blobs))} == {b"gamma"}
 
 
+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"}
+    )
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    table = db.create_table("t", schema=schema)
+    table.add([{"id": 1, "image": b"frag-one"}])
+    table.add([{"id": 2, "image": b"frag-two"}])
+    by_id = _row_ids_by_id(table)
+    ids = [by_id[1], by_id[2]]
+
+    table.optimize()
+
+    blobs = table.fetch_blobs("image", ids)
+    assert blobs.to_pylist() == [b"frag-one", b"frag-two"]
+    ranges = table.fetch_blob_ranges("image", [(ids[0], 5, 3), (ids[1], 5, 3)])
+    assert ranges.to_pylist() == [b"one", b"two"]
+
+
 def test_fetch_blobs_preserves_null_and_empty_values():
     table = _blob_table(
         "nulls",
@@ -739,8 +777,25 @@ def test_fetch_blob_ranges_validates_requests():
     with pytest.raises(ValueError, match="offset \\+ length overflowed"):
         table.fetch_blob_ranges("image", [(row_id, 2**64 - 1, 1)])
 
-    with pytest.raises(ValueError, match="row IDs"):
+    with pytest.raises(ValueError) as exc_info:
         table.fetch_blob_ranges("image", [(2**64 - 1, 0, 1)])
+    _assert_missing_blob_row_ids(exc_info)
+
+
+def test_fetch_blob_apis_reject_missing_fragment_row_addr():
+    table = _blob_table("missing_frag", [{"id": 1, "image": b"x"}])
+    live = _row_ids_by_id(table)[1]
+    _assert_fetch_apis_reject_missing_row_ids(table, [1 << 32, live])
+
+
+def test_fetch_blob_apis_reject_deleted_row_ids():
+    table = _blob_table(
+        "deleted_rows",
+        [{"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():
diff --git a/rust/lancedb/src/blob.rs b/rust/lancedb/src/blob.rs
index d59123ec3..6a0d968b0 100644
--- a/rust/lancedb/src/blob.rs
+++ b/rust/lancedb/src/blob.rs
@@ -7,7 +7,9 @@
 //! raw `Binary` / `LargeBinary` into the blob struct layout. Queries return
 //! small descriptors, not bytes.
 //!
-//! Blob tables require Lance file format >= 2.2 and stable row ids at create.
+//! Blob tables require Lance file format >= 2.2. `_rowid` values stay valid
+//! after compaction when the table has stable row ids. Overwrite is a new
+//! create and does not keep the previous table's stable row id setting.
 
 use std::ops::Range;
 use std::sync::Arc;
@@ -324,6 +326,7 @@ pub(crate) fn blob_column_names(schema: &Schema) -> Vec<String> {
 }
 
 /// Bumps storage format to at least [`LanceFileVersion::V2_2`] for blob schemas.
+/// Leaves `enable_stable_row_ids` unchanged.
 pub(crate) fn ensure_blob_storage_version(schema: &Schema, params: &mut WriteParams) {
     if !has_blob_columns(schema) {
         return;
@@ -385,6 +388,30 @@ fn ensure_all_row_ids_resolved(column: &str, requested: usize, resolved: usize)
     }
 }
 
+/// 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,
+    };
+
+    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"
+            ),
+        }
+    } else {
+        err.into()
+    }
+}
+
 /// Materialize blob-local ranges (same length and order as `requests`, nulls preserved).
 pub(crate) async fn take_blob_ranges_aligned(
     dataset: &Arc<Dataset>,
@@ -405,7 +432,8 @@ pub(crate) async fn take_blob_ranges_aligned(
         .with_row_ids(lance_requests)
         .preserve_order(true)
         .execute()
-        .await?;
+        .await
+        .map_err(|err| map_blob_take_error(column, requests.len(), err))?;
     ensure_all_row_ids_resolved(column, requests.len(), payloads.len())?;
 
     let mut builder = LargeBinaryBuilder::new();
@@ -434,7 +462,8 @@ pub(crate) async fn take_blobs_aligned(
         .with_row_ids(row_ids.to_vec())
         .preserve_order(true)
         .execute()
-        .await?;
+        .await
+        .map_err(|err| map_blob_take_error(column, row_ids.len(), err))?;
     ensure_all_row_ids_resolved(column, row_ids.len(), payloads.len())?;
 
     let mut builder = LargeBinaryBuilder::new();
@@ -458,7 +487,10 @@ pub(crate) async fn take_blob_files_aligned(
         return Ok(Vec::new());
     }
 
-    let handles = dataset.take_blobs(row_ids, column).await?;
+    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())?;
     Ok(handles
         .into_iter()
@@ -504,6 +536,21 @@ mod tests {
             params.data_storage_version.unwrap().resolve(),
             ConcreteFileVersion::V2_2
         );
+        assert!(!params.enable_stable_row_ids);
+    }
+
+    #[test]
+    fn storage_version_leaves_stable_row_ids_enabled() {
+        let mut params = WriteParams {
+            enable_stable_row_ids: true,
+            ..Default::default()
+        };
+        ensure_blob_storage_version(&blob_schema(), &mut params);
+        assert!(params.enable_stable_row_ids);
+        assert_eq!(
+            params.data_storage_version.unwrap().resolve(),
+            ConcreteFileVersion::V2_2
+        );
     }
 
     #[test]
@@ -576,5 +623,6 @@ mod tests {
         let mut params = WriteParams::default();
         ensure_blob_storage_version(&schema, &mut params);
         assert!(params.data_storage_version.is_none());
+        assert!(!params.enable_stable_row_ids);
     }
 }
diff --git a/rust/lancedb/src/database/listing.rs b/rust/lancedb/src/database/listing.rs
index c22b73dd7..c6d834c5a 100644
--- a/rust/lancedb/src/database/listing.rs
+++ b/rust/lancedb/src/database/listing.rs
@@ -18,7 +18,7 @@ use lance_table::io::commit::commit_handler_from_url;
 use object_store::local::LocalFileSystem;
 use snafu::ResultExt;
 
-use crate::blob::{ensure_blob_storage_version, has_blob_columns};
+use crate::blob::ensure_blob_storage_version;
 use crate::connection::ConnectRequest;
 use crate::database::ReadConsistency;
 use crate::database::namespace::LanceNamespaceDatabase;
@@ -827,7 +827,6 @@ impl ListingDatabase {
         if let Some(enable_stable_row_ids) = overrides
             .enable_stable_row_ids
             .or(self.new_table_config.enable_stable_row_ids)
-            .or(has_blob_columns(&data_schema).then_some(true))
         {
             write_params.enable_stable_row_ids = enable_stable_row_ids;
         }
diff --git a/rust/lancedb/src/database/namespace.rs b/rust/lancedb/src/database/namespace.rs
index 5ca720e85..78641a8b3 100644
--- a/rust/lancedb/src/database/namespace.rs
+++ b/rust/lancedb/src/database/namespace.rs
@@ -23,7 +23,7 @@ use lance_namespace_impls::ConnectBuilder;
 use lance_table::io::commit::CommitHandler;
 use lance_table::io::commit::external_manifest::ExternalManifestCommitHandler;
 
-use crate::blob::{ensure_blob_storage_version, has_blob_columns};
+use crate::blob::ensure_blob_storage_version;
 use crate::connection::NamespaceClientPushdownOperation;
 use crate::database::ReadConsistency;
 use crate::database::listing::{NewTableConfig, take_request_creation_overrides};
@@ -217,7 +217,6 @@ impl LanceNamespaceDatabase {
         if let Some(enable_stable_row_ids) = overrides
             .enable_stable_row_ids
             .or(self.new_table_config.enable_stable_row_ids)
-            .or(has_blob_columns(data_schema.as_ref()).then_some(true))
         {
             params.enable_stable_row_ids = enable_stable_row_ids;
         }
diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs
index 33b6ea8ce..44e12d8ad 100644
--- a/rust/lancedb/src/table.rs
+++ b/rust/lancedb/src/table.rs
@@ -1192,6 +1192,9 @@ impl Table {
     /// valid empty blobs contain empty byte strings. Prefer
     /// [`Self::fetch_blob_files`] for large selections.
     ///
+    /// `_rowid` values stay valid after compaction when the table has stable
+    /// row ids.
+    ///
     /// ```
     /// use arrow_array::UInt64Array;
     /// use futures::TryStreamExt;
@@ -1233,6 +1236,9 @@ impl Table {
     /// the requests. Null blobs produce null output slots; empty ranges on
     /// non-null blobs produce empty byte strings.
     ///
+    /// `_rowid` values stay valid after compaction when the table has stable
+    /// row ids.
+    ///
     /// ```
     /// use lancedb::blob::BlobRangeRequest;
     ///
@@ -1271,6 +1277,9 @@ impl Table {
     /// Same length and order as `row_ids`. Null rows are `None`. Bytes are not
     /// read from disk until a call to [`BlobFile::read`].
     ///
+    /// `_rowid` values stay valid after compaction when the table has stable
+    /// row ids.
+    ///
     /// ```
     /// # use lancedb::Table;
     /// # async fn lazy_read(table: &Table, row_ids: &[u64]) -> Result<(), Box<dyn std::error::Error>> {
diff --git a/rust/lancedb/tests/blob_integration.rs b/rust/lancedb/tests/blob_integration.rs
index 7b709b645..b884a48f7 100644
--- a/rust/lancedb/tests/blob_integration.rs
+++ b/rust/lancedb/tests/blob_integration.rs
@@ -111,7 +111,7 @@ async fn query_image_struct(table: &Table) -> StructArray {
 }
 
 #[tokio::test]
-async fn declaring_blob_column_bumps_format_and_enables_stable_row_ids() -> Result<()> {
+async fn declaring_blob_column_uses_v2_2_and_default_row_ids() -> Result<()> {
     let tmp = tempdir().unwrap();
     let db = connect(tmp.path().to_str().unwrap()).execute().await?;
     let table = db
@@ -120,12 +120,12 @@ async fn declaring_blob_column_bumps_format_and_enables_stable_row_ids() -> Resu
         .await?;
 
     assert!(supports_blob_v2(storage_format_version(&table).await));
-    assert!(uses_stable_row_ids(&table).await);
+    assert!(!uses_stable_row_ids(&table).await);
     Ok(())
 }
 
 #[tokio::test]
-async fn explicit_stable_row_id_setting_wins_over_blob_default() -> Result<()> {
+async fn blob_create_honors_disabled_stable_row_ids() -> Result<()> {
     let tmp = tempdir().unwrap();
     let db = connect(tmp.path().to_str().unwrap()).execute().await?;
     let table = db
@@ -179,7 +179,7 @@ async fn creating_with_blob_data_bumps_format() -> Result<()> {
     let table = db.create_table("t", batch).execute().await?;
 
     assert!(supports_blob_v2(storage_format_version(&table).await));
-    assert!(uses_stable_row_ids(&table).await);
+    assert!(!uses_stable_row_ids(&table).await);
     assert_eq!(table.count_rows(None).await?, 1);
     Ok(())
 }
@@ -277,7 +277,7 @@ async fn add_rejects_uncoercible_blob_input() -> Result<()> {
 }
 
 #[tokio::test]
-async fn connection_level_stable_row_id_setting_wins_over_blob_default() -> Result<()> {
+async fn connection_disables_stable_row_ids_on_blob_create() -> Result<()> {
     let tmp = tempdir().unwrap();
     let db = connect(tmp.path().to_str().unwrap())
         .storage_option(OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, "false")
@@ -294,7 +294,7 @@ async fn connection_level_stable_row_id_setting_wins_over_blob_default() -> Resu
 }
 
 #[tokio::test]
-async fn namespace_create_applies_blob_defaults() -> Result<()> {
+async fn namespace_blob_create_uses_v2_2_and_default_row_ids() -> Result<()> {
     let tmp = tempdir().unwrap();
     let mut properties = std::collections::HashMap::new();
     properties.insert("root".to_string(), tmp.path().to_str().unwrap().to_string());
@@ -304,6 +304,23 @@ async fn namespace_create_applies_blob_defaults() -> Result<()> {
         .execute()
         .await?;
 
+    assert!(supports_blob_v2(storage_format_version(&table).await));
+    assert!(!uses_stable_row_ids(&table).await);
+    Ok(())
+}
+
+#[tokio::test]
+async fn namespace_create_honors_enabled_stable_row_ids() -> Result<()> {
+    let tmp = tempdir().unwrap();
+    let mut properties = std::collections::HashMap::new();
+    properties.insert("root".to_string(), tmp.path().to_str().unwrap().to_string());
+    let db = connect_namespace("dir", properties).execute().await?;
+    let table = db
+        .create_empty_table("t", blob_table_schema())
+        .storage_option(OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, "true")
+        .execute()
+        .await?;
+
     assert!(supports_blob_v2(storage_format_version(&table).await));
     assert!(uses_stable_row_ids(&table).await);
     Ok(())
@@ -430,6 +447,35 @@ async fn collect_id_rowid(table: &Table) -> Result<Vec<(i64, u64)>> {
         .collect())
 }
 
+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("rowaddr"), "{message}");
+    assert!(!message.contains("fragment"), "{message}");
+}
+
+async fn assert_fetch_apis_reject_missing_row_ids(table: &Table, row_ids: &[u64]) -> Result<()> {
+    let err = table.fetch_blobs("image", row_ids).await.unwrap_err();
+    assert_missing_blob_row_ids(&err);
+
+    let err = table.fetch_blob_files("image", row_ids).await.unwrap_err();
+    assert_missing_blob_row_ids(&err);
+
+    let err = table
+        .fetch_blob_ranges(
+            "image",
+            row_ids
+                .iter()
+                .copied()
+                .map(|row_id| BlobRangeRequest::new(row_id, 0, 1)),
+        )
+        .await
+        .unwrap_err();
+    assert_missing_blob_row_ids(&err);
+    Ok(())
+}
+
 #[tokio::test]
 async fn fetch_blobs_round_trips_bytes() -> Result<()> {
     let tmp = tempdir().unwrap();
@@ -482,7 +528,7 @@ async fn fetch_blobs_round_trips_nested_blob_column() -> Result<()> {
     let table = db.create_table("t", batch).execute().await?;
 
     assert!(supports_blob_v2(storage_format_version(&table).await));
-    assert!(uses_stable_row_ids(&table).await);
+    assert!(!uses_stable_row_ids(&table).await);
 
     let ids = collect_row_ids(&table).await?;
     let bytes = table.fetch_blobs("info.blob", &ids).await?;
@@ -656,8 +702,7 @@ async fn fetch_blob_ranges_validates_requests() -> Result<()> {
         .fetch_blob_ranges("image", [BlobRangeRequest::new(u64::MAX, 0, 1)])
         .await
         .unwrap_err();
-    assert!(matches!(&err, Error::InvalidInput { .. }), "got {err:?}");
-    assert!(err.to_string().contains("row IDs"));
+    assert_missing_blob_row_ids(&err);
     Ok(())
 }
 
@@ -690,7 +735,21 @@ async fn fetch_blobs_out_of_range_id_errors_without_panic() -> Result<()> {
     let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"x".as_slice())]).await?;
 
     let err = table.fetch_blobs("image", &[u64::MAX]).await.unwrap_err();
-    assert!(err.to_string().contains("row IDs"));
+    assert_missing_blob_row_ids(&err);
+    Ok(())
+}
+
+#[tokio::test]
+async fn fetch_blob_files_rejects_missing_fragment_row_addr() -> Result<()> {
+    let tmp = tempdir().unwrap();
+    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
+    let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"x".as_slice())]).await?;
+
+    let err = table
+        .fetch_blob_files("image", &[1u64 << 32])
+        .await
+        .unwrap_err();
+    assert_missing_blob_row_ids(&err);
     Ok(())
 }
 
@@ -700,24 +759,25 @@ async fn fetch_blob_apis_reject_mixed_valid_and_missing_row_ids() -> Result<()>
     let db = connect(tmp.path().to_str().unwrap()).execute().await?;
     let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"x".as_slice())]).await?;
     let row_id = collect_row_ids(&table).await?[0];
-    let row_ids = [u64::MAX, row_id];
+    let missing_row_addr = 1u64 << 32;
+    let row_ids = [missing_row_addr, row_id];
+    assert_fetch_apis_reject_missing_row_ids(&table, &row_ids).await
+}
 
-    let err = table.fetch_blobs("image", &row_ids).await.unwrap_err();
-    assert!(matches!(&err, Error::InvalidInput { .. }), "got {err:?}");
-    assert!(err.to_string().contains("row IDs"));
+#[tokio::test]
+async fn fetch_blob_apis_reject_deleted_row_ids() -> 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"one".as_slice()), Some(b"two")])
+            .await?;
+    let pairs = collect_id_rowid(&table).await?;
+    let deleted_row_addr = pairs.iter().find(|(id, _)| *id == 2).unwrap().1;
+    let live_row_addr = pairs.iter().find(|(id, _)| *id == 1).unwrap().1;
 
-    let err = table.fetch_blob_files("image", &row_ids).await.unwrap_err();
-    assert!(matches!(&err, Error::InvalidInput { .. }), "got {err:?}");
-    assert!(err.to_string().contains("row IDs"));
+    table.delete("id = 2").await?;
 
-    let requests = row_ids.map(|row_id| BlobRangeRequest::new(row_id, 0, 1));
-    let err = table
-        .fetch_blob_ranges("image", requests)
-        .await
-        .unwrap_err();
-    assert!(matches!(&err, Error::InvalidInput { .. }), "got {err:?}");
-    assert!(err.to_string().contains("row IDs"));
-    Ok(())
+    assert_fetch_apis_reject_missing_row_ids(&table, &[deleted_row_addr, live_row_addr]).await
 }
 
 #[tokio::test]
@@ -920,7 +980,10 @@ async fn fetch_blobs_after_delete() -> Result<()> {
 #[tokio::test]
 async fn fetch_blobs_with_precompaction_row_ids_survives_compaction() -> Result<()> {
     let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
+    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], &[Some(b"frag-one".as_slice())]).await?;
     table
         .add(binary_input_batch(&[2], &[Some(b"frag-two".as_slice())]))