feat(blob): reuse independently addressed managed objects

docs/src/guide/blob.md

@@ -61,6 +61,29 @@ source of truth for which scheme is supported at each `data_storage_version`.
 | `0.1`, `2.0`, `2.1` | Supported for write/read | Not supported |
 | `2.2+` | Not supported for write | Supported for write/read (recommended) |
 
+### Managed objects and client compatibility
+
+Current writers store out-of-line Blob v2 payloads in independently named
+`_blobs/<uuid>.blob` objects and publish the Managed Blob reader and writer
+capability on the table. This works with file formats 2.2 and 2.3; it does not
+require choosing 2.3. Clients that do not understand the capability must refuse
+to open a flagged snapshot. Updating an existing table with Blob data files can
+activate it, even if that batch contains only inline values. The capability
+remains set across later writes and restores.
+
+Compaction preserves Managed payload objects and can adopt existing Packed or
+Dedicated sidecars in place. It records their complete addresses, so deleting
+the original data file does not require copying its sidecars. Cleanup retains
+objects referenced by protected snapshots; a partially live packed object is
+retained as a whole. Blob reads continue to return the same bytes, while raw
+descriptor scans can now report `kind = 4` for Managed values.
+
+After activation, use a client that supports Managed Blobs for all table
+maintenance. Older clients, including v11.0.0, can bypass capability checks when
+running cleanup from an unflagged historical snapshot or a cached handle. Such
+cleanup can delete adopted sidecars by treating their original data file as
+their owner. The table flag does not retrofit those old maintenance paths.
+
 ## Blob v2: Write Patterns
 
 Use `blob_field` and `blob_array` to build blob v2 columns.

python/python/tests/test_blob.py

@@ -138,7 +138,7 @@ def _add_columns_blob_v2_values(tmp_path):
 def _assert_blob_v2_add_columns_result(dataset, column, payloads):
     desc = dataset.to_table(columns=[column]).column(column).chunk(0)
 
-    assert desc.field("kind").to_pylist() == [0, 1, 2, 3]
+    assert desc.field("kind").to_pylist() == [0, 4, 4, 3]
     assert desc.field("blob_id").to_pylist()[3] == 1
     assert desc.field("blob_uri").to_pylist()[3] == "external_blob.bin"
 
@@ -1411,7 +1411,7 @@ def test_blob_extension_inline_threshold_per_column(tmp_path):
 
     desc = ds.to_table(columns=["inline_blob", "packed_blob"])
     assert desc.column("inline_blob").chunk(0).field("kind").to_pylist() == [0]
-    assert desc.column("packed_blob").chunk(0).field("kind").to_pylist() == [1]
+    assert desc.column("packed_blob").chunk(0).field("kind").to_pylist() == [4]
 
 
 def test_blob_extension_threshold_metadata_persists_after_reopen(tmp_path):
@@ -1514,7 +1514,7 @@ def test_blob_extension_dedicated_threshold_precedes_inline_threshold(tmp_path):
     )
 
     desc = ds.to_table(columns=["blob"]).column("blob").chunk(0)
-    assert desc.field("kind").to_pylist() == [2]
+    assert desc.field("kind").to_pylist() == [4]
 
 
 def test_blob_extension_write_external(tmp_path):
@@ -1655,6 +1655,15 @@ def failing_reader():
         ds.add_columns(failing_reader(), reader_schema=schema)
 
     assert ds.version == 1
+    files_after = _dataset_file_set(dataset_path)
+    assert files_before <= files_after
+    orphans = files_after - files_before
+    assert orphans and all(
+        p.parts[0] == "_blobs" and p.suffix == ".blob" for p in orphans
+    )
+    # Independent payloads from failed writes follow the existing orphan policy.
+    # No concurrent writer is running here, so immediate unverified GC is safe.
+    ds.cleanup_old_versions(delete_unverified=True)
     assert _dataset_file_set(dataset_path) == files_before
     assert external_blob_path.exists()
 
@@ -1685,6 +1694,15 @@ def fail_on_second_fragment(batch):
 
     assert call_count == 2
     assert ds.version == 1
+    files_after = _dataset_file_set(dataset_path)
+    assert files_before <= files_after
+    orphans = files_after - files_before
+    assert orphans and all(
+        p.parts[0] == "_blobs" and p.suffix == ".blob" for p in orphans
+    )
+    # Independent payloads from failed writes follow the existing orphan policy.
+    # No concurrent writer is running here, so immediate unverified GC is safe.
+    ds.cleanup_old_versions(delete_unverified=True)
     assert _dataset_file_set(dataset_path) == files_before
     assert external_blob_path.exists()
 
@@ -2453,7 +2471,7 @@ def test_blob_v2_lazy_preserves_empty_and_null(tmp_path, values, has_sidecar):
         assert descriptions[0]["size"] == 0
     if has_sidecar:
         assert any(
-            description is not None and description["kind"] == 1
+            description is not None and description["kind"] == 4
             for description in descriptions
         )
         assert any(path.suffix == ".blob" for path in _dataset_file_set(dataset_path))
@@ -2712,7 +2730,7 @@ def test_write_nested_blob_v2_and_take_by_field_path(tmp_path):
     )
 
     desc = dataset.to_table(columns=["info.blob"]).column("info.blob").chunk(0)
-    assert desc.field("kind").to_pylist()[:2] == [0, 1]
+    assert desc.field("kind").to_pylist()[:2] == [0, 4]
 
     blobs = dataset.take_blobs("info.blob", indices=[0, 1])
     with blobs[0] as f:

rust/lance-core/src/datatypes.rs

@@ -596,6 +596,12 @@ pub enum BlobKind {
     /// External blobs can have a position and a size. If the position is not set,
     /// it defaults to 0, which points to the beginning of the blob.
     External = 3,
+    /// A Lance-owned immutable object, independently of the descriptor's data file.
+    /// `blob_id` is the exact manifest base ID (including zero); `blob_uri` is
+    /// relative to that base root. `position`/`size` select a known range, and
+    /// zero size is an empty value rather than a request to discover its length.
+    /// Tables containing this kind require the Managed Blob reader and writer feature flags.
+    Managed = 4,
 }
 
 impl TryFrom<u8> for BlobKind {
@@ -607,6 +613,7 @@ impl TryFrom<u8> for BlobKind {
             1 => Ok(Self::Packed),
             2 => Ok(Self::Dedicated),
             3 => Ok(Self::External),
+            4 => Ok(Self::Managed),
             other => Err(Error::invalid_input_source(
                 format!("Unknown blob kind {other:?}").into(),
             )),

rust/lance-core/src/utils/blob.rs

@@ -3,6 +3,35 @@
 
 use object_store::path::Path;
 
+use crate::{Error, Result};
+
+/// Validate a Managed descriptor's object-relative path and known byte range.
+///
+/// No base ID is reserved. Resolving the base belongs to the snapshot holding
+/// the descriptor, and must fail if that snapshot has no matching binding.
+pub fn validate_managed_reference(uri: &str, position: u64, size: u64) -> Result<Path> {
+    if uri.is_empty()
+        || uri.starts_with('/')
+        || uri.ends_with('/')
+        || uri.contains("://")
+        || uri.contains('\\')
+        || uri
+            .split('/')
+            .any(|part| part.is_empty() || part == "." || part == "..")
+    {
+        return Err(Error::invalid_input(format!(
+            "Managed blob_uri must be a non-empty canonical relative object path, got {uri:?}"
+        )));
+    }
+    position.checked_add(size).ok_or_else(|| {
+        Error::invalid_input(format!(
+            "Managed blob range overflows u64: position={position}, size={size}"
+        ))
+    })?;
+    Path::parse(uri)
+        .map_err(|error| Error::invalid_input(format!("Invalid Managed blob_uri {uri:?}: {error}")))
+}
+
 /// Format a blob sidecar path for a data file.
 ///
 /// Layout: `<base>/<data_file_key>/<obfuscated_blob_id>.blob`
@@ -17,6 +46,34 @@ pub fn blob_path(base: &Path, data_file_key: &str, blob_id: u32) -> Path {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use rstest::rstest;
+
+    #[rstest]
+    #[case("")]
+    #[case("/data/a.blob")]
+    #[case("data/../a.blob")]
+    #[case("data/./a.blob")]
+    #[case("data//a.blob")]
+    #[case("data/a.blob/")]
+    #[case("s3://bucket/a.blob")]
+    #[case("data\\a.blob")]
+    fn managed_paths_reject_noncanonical_references(#[case] uri: &str) {
+        let error = validate_managed_reference(uri, 0, 1).unwrap_err();
+        assert!(matches!(error, Error::InvalidInput { .. }));
+        assert!(error.to_string().contains("Managed blob_uri"));
+    }
+
+    #[test]
+    fn managed_ranges_preserve_empty_values_and_reject_overflow() {
+        assert!(validate_managed_reference("data/a.blob", u64::MAX, 0).is_ok());
+        let error = validate_managed_reference("data/a.blob", u64::MAX, 1).unwrap_err();
+        assert!(matches!(error, Error::InvalidInput { .. }));
+        assert!(
+            error
+                .to_string()
+                .contains("position=18446744073709551615, size=1")
+        );
+    }
 
     #[test]
     fn test_blob_path_formatting() {

rust/lance-encoding/src/encoder/structural.rs

@@ -72,7 +72,10 @@ impl PrimitiveFieldEncoding {
         }
     }
 
-    fn create_at(
+    /// Compose a primitive encoder at an already allocated physical column.
+    /// File grammars use this when another logical type, such as a Blob,
+    /// supplies the physical descriptor field instead of the logical field.
+    pub fn create_at(
         &self,
         field: Field,
         column_index: u32,

rust/lance-encoding/src/encodings/logical/blob.rs

@@ -329,67 +329,91 @@ impl FieldEncoder for BlobV2StructuralEncoder {
         let mut uri_builder = StringBuilder::with_capacity(row_count, row_count * 16);
 
         for i in 0..row_count {
-            let (kind_value, position_value, size_value, blob_id_value, uri_value) =
-                if struct_arr.is_null(i) || kind_col.is_null(i) {
-                    (BlobKind::Inline as u8, 0, 0, 0, "".to_string())
-                } else {
-                    let kind_val = BlobKind::try_from(kind_col.value(i))?;
-                    match kind_val {
-                        BlobKind::Dedicated => (
-                            BlobKind::Dedicated as u8,
-                            0,
-                            blob_size_col.value(i),
-                            blob_id_col.value(i),
-                            "".to_string(),
-                        ),
-                        BlobKind::External => {
-                            let uri = uri_col.value(i).to_string();
-                            let position = if packed_position_col.is_null(i) {
-                                0
-                            } else {
-                                packed_position_col.value(i)
-                            };
-                            let size = if blob_size_col.is_null(i) {
-                                0
-                            } else {
-                                blob_size_col.value(i)
-                            };
-                            let external_base_id = if blob_id_col.is_null(i) {
-                                0
-                            } else {
-                                blob_id_col.value(i)
-                            };
-                            (
-                                BlobKind::External as u8,
-                                position,
-                                size,
-                                external_base_id,
-                                uri,
-                            )
+            let (kind_value, position_value, size_value, blob_id_value, uri_value) = if struct_arr
+                .is_null(i)
+                || kind_col.is_null(i)
+            {
+                (BlobKind::Inline as u8, 0, 0, 0, "".to_string())
+            } else {
+                let kind_val = BlobKind::try_from(kind_col.value(i))?;
+                match kind_val {
+                    BlobKind::Managed => {
+                        if uri_col.is_null(i)
+                            || blob_id_col.is_null(i)
+                            || packed_position_col.is_null(i)
+                            || blob_size_col.is_null(i)
+                        {
+                            return Err(Error::invalid_input(format!(
+                                "Managed blob row {i} requires URI, base ID, position, and size"
+                            )));
                         }
-                        BlobKind::Packed => (
-                            BlobKind::Packed as u8,
-                            packed_position_col.value(i),
-                            blob_size_col.value(i),
+                        let uri = uri_col.value(i);
+                        let position = packed_position_col.value(i);
+                        let size = blob_size_col.value(i);
+                        lance_core::utils::blob::validate_managed_reference(uri, position, size)?;
+                        (
+                            BlobKind::Managed as u8,
+                            position,
+                            size,
                             blob_id_col.value(i),
+                            uri.to_string(),
+                        )
+                    }
+                    BlobKind::Dedicated => (
+                        BlobKind::Dedicated as u8,
+                        0,
+                        blob_size_col.value(i),
+                        blob_id_col.value(i),
+                        "".to_string(),
+                    ),
+                    BlobKind::External => {
+                        let uri = uri_col.value(i).to_string();
+                        let position = if packed_position_col.is_null(i) {
+                            0
+                        } else {
+                            packed_position_col.value(i)
+                        };
+                        let size = if blob_size_col.is_null(i) {
+                            0
+                        } else {
+                            blob_size_col.value(i)
+                        };
+                        let external_base_id = if blob_id_col.is_null(i) {
+                            0
+                        } else {
+                            blob_id_col.value(i)
+                        };
+                        (
+                            BlobKind::External as u8,
+                            position,
+                            size,
+                            external_base_id,
+                            uri,
+                        )
+                    }
+                    BlobKind::Packed => (
+                        BlobKind::Packed as u8,
+                        packed_position_col.value(i),
+                        blob_size_col.value(i),
+                        blob_id_col.value(i),
+                        "".to_string(),
+                    ),
+                    BlobKind::Inline => {
+                        let data_val = data_col.value(i);
+                        let blob_len = data_val.len() as u64;
+                        let position =
+                            external_buffers.add_buffer(LanceBuffer::from(Buffer::from(data_val)));
+
+                        (
+                            BlobKind::Inline as u8,
+                            position,
+                            blob_len,
+                            0,
                             "".to_string(),
-                        ),
-                        BlobKind::Inline => {
-                            let data_val = data_col.value(i);
-                            let blob_len = data_val.len() as u64;
-                            let position = external_buffers
-                                .add_buffer(LanceBuffer::from(Buffer::from(data_val)));
-
-                            (
-                                BlobKind::Inline as u8,
-                                position,
-                                blob_len,
-                                0,
-                                "".to_string(),
-                            )
-                        }
+                        )
                     }
-                };
+                }
+            };
 
             kind_builder.append_value(kind_value);
             position_builder.append_value(position_value);

rust/lance-file/src/concat.rs

@@ -437,6 +437,22 @@ fn validate_blob_field(
                         ));
                     }
                 }
+                BlobKind::Managed => {
+                    let uris = descriptors
+                        .column_by_name("blob_uri")
+                        .ok_or_else(|| {
+                            Error::corrupt_file(path.clone(), "Managed descriptor has no blob_uri")
+                        })?
+                        .as_string::<i32>();
+                    lance_core::utils::blob::validate_managed_reference(
+                        uris.value(row),
+                        positions.value(row),
+                        sizes.value(row),
+                    )?;
+                    // The ID is a snapshot base binding, not a leased sidecar
+                    // number. Concatenation preserves the independent object
+                    // address; the dataset caller owns the base namespace.
+                }
                 BlobKind::Inline | BlobKind::External => {}
             }
         }