feat(blob): expose managed sharing in Python preparation API

docs/src/guide/blob.md

@@ -125,6 +125,55 @@ inspect. Such cleanup can delete adopted sidecars that newer snapshots still
 reference, because it treats the original data file as their owner. The table
 flag therefore does not protect against these old maintenance paths.
 
+### Explicit sharing within a table
+
+Advanced writers can reuse a Managed payload through
+`BlobDescriptorArrayBuilder.append_managed(uri, offset, size, base_id=None)`.
+The builder validates the relative path and range without uploading or copying
+bytes. The new reference becomes live only when the dataset write commits.
+For example, write one payload and append two references to the same bytes:
+
+```python
+import lance
+import pyarrow as pa
+
+schema = pa.schema([lance.blob_field("blob", inline_size_threshold=0)])
+ds = lance.write_dataset(
+    pa.Table.from_arrays([lance.blob_array([b"shared"])], schema=schema),
+    "./shared_blobs.lance",
+    data_storage_version="2.2",
+)
+descriptor = ds.to_table(blob_handling="all_descriptions")["blob"][0].as_py()
+assert descriptor["kind"] == 4  # Relative to this table's primary base.
+
+builder = lance.BlobDescriptorArrayBuilder("blob")
+for _ in range(2):
+    builder.append_managed(
+        descriptor["blob_uri"], descriptor["position"], descriptor["size"]
+    )
+references = pa.Table.from_arrays(
+    [builder.finish()], schema=pa.schema([builder.field])
+)
+ds = lance.write_dataset(references, ds.uri, mode="append")
+assert [data for _, data in ds.read_blobs("blob", indices=range(3))] == [
+    b"shared", b"shared", b"shared"
+]
+```
+
+The three rows share one immutable payload object, including after compaction.
+Cleanup retains the whole object while any retained snapshot references it.
+Keep the source snapshot protected, for example with a tag, until the append
+commits if concurrent updates and cleanup can otherwise retire it. An open
+dataset handle alone does not protect a snapshot from cleanup.
+
+Use references from the same table and preserve their base identity. Kind 5
+descriptors require their registered `blob_id` as `base_id`; for kind 4, the base
+comes from the source data file. The example uses newly written data in the
+table's primary base. Do not copy an implicit reference from a shallow clone's
+source file into a destination with a different base. Cross-table sharing and
+automatic deduplication are outside this API's contract. All maintenance clients
+must support Managed Blobs as described above.
+
 ## Blob v2: Write Patterns
 
 Use `blob_field` and `blob_array` to build blob v2 columns.

python/python/lance/lance/__init__.pyi

@@ -235,6 +235,9 @@ class BlobDescriptorArrayBuilder:
         self, blob_id: int, offsets: List[int], sizes: List[int]
     ) -> None: ...
     def append_dedicated(self, blob_id: int, size: int) -> None: ...
+    def append_managed(
+        self, uri: str, offset: int, size: int, *, base_id: Optional[int] = None
+    ) -> None: ...
     def append(self, value: BlobDescriptor) -> None: ...
     def extend(self, values: List[BlobDescriptor]) -> None: ...
     def append_inline(self, data: bytes) -> None: ...

python/python/tests/test_blob.py

@@ -1991,6 +1991,117 @@ def test_blob_extension_add_columns_all_nulls_blob_v2(tmp_path):
     assert [data for _, _, data in ranges] == [None] * 4
 
 
+@pytest.mark.parametrize("version", ["2.2", "2.3"])
+@pytest.mark.parametrize("nested", [False, True], ids=["top_level", "struct"])
+def test_prepared_managed_blob_sharing_lifecycle(tmp_path, version, nested):
+    dataset_uri = tmp_path / "shared.lance"
+
+    def table(ids, field, values):
+        if nested:
+            values = pa.StructArray.from_arrays([values], fields=[field])
+            field = pa.field(
+                "value", values.type, metadata={b"example:type": b"opaque"}
+            )
+        return pa.Table.from_arrays(
+            [pa.array(ids, type=pa.int64()), values],
+            schema=pa.schema([pa.field("id", pa.int64()), field]),
+        )
+
+    initial = table(
+        [0],
+        lance.blob_field("blob", inline_size_threshold=0),
+        lance.blob_array([b"shared"]),
+    )
+    ds = lance.write_dataset(initial, dataset_uri, data_storage_version=version)
+    blob_column = "value.blob" if nested else "blob"
+    row = ds.to_table(blob_handling="all_descriptions").to_pylist()[0]
+    descriptor = (row["value"] if nested else row)["blob"]
+    assert descriptor["kind"] == 4
+    payload_files = set(dataset_uri.glob("_blobs/*.blob"))
+    assert len(payload_files) == 1
+
+    builder = lance.BlobDescriptorArrayBuilder("blob")
+    builder.append_managed(
+        descriptor["blob_uri"], descriptor["position"], descriptor["size"]
+    )
+    builder.append_managed(descriptor["blob_uri"], descriptor["position"] + 1, 3)
+    builder.append_null()
+    ds = lance.write_dataset(
+        table([1, 2, 3], builder.field, builder.finish()), dataset_uri, mode="append"
+    )
+    assert len(ds.get_fragments()) == 2
+    assert [data for _, data in ds.read_blobs(blob_column, indices=range(4))] == [
+        b"shared",
+        b"shared",
+        b"har",
+        None,
+    ]
+    assert set(dataset_uri.glob("_blobs/*.blob")) == payload_files
+
+    ds.optimize.compact_files()
+    assert len(ds.get_fragments()) == 1
+    ds.delete("id = 0")
+    ds.cleanup_old_versions(retain_versions=1, delete_unverified=True)
+    assert [data for _, data in ds.read_blobs(blob_column, indices=range(3))] == [
+        b"shared",
+        b"har",
+        None,
+    ]
+    assert set(dataset_uri.glob("_blobs/*.blob")) == payload_files
+    if nested:
+        assert ds.schema.field("value").metadata == {b"example:type": b"opaque"}
+
+    protected_version = ds.version
+    ds.tags.create("shared", protected_version)
+    ds.delete("id < 3")
+    ds.cleanup_old_versions(
+        retain_versions=1, delete_unverified=True, error_if_tagged_old_versions=False
+    )
+    protected = lance.dataset(dataset_uri, version=protected_version)
+    assert protected.take_blobs(blob_column, indices=[0])[0].readall() == b"shared"
+    assert set(dataset_uri.glob("_blobs/*.blob")) == payload_files
+
+    ds.tags.delete("shared")
+    ds.cleanup_old_versions(retain_versions=1, delete_unverified=True)
+    assert not list(dataset_uri.glob("_blobs/*.blob"))
+    assert ds.take_blobs(blob_column, indices=[0]) == [None]
+
+
+@pytest.mark.parametrize("base_id", [None, 7])
+def test_blob_descriptor_array_builder_managed(base_id):
+    builder = lance.BlobDescriptorArrayBuilder("blob")
+    builder.append_managed("_blobs/existing.blob", 10, 0, base_id=base_id)
+    assert builder.finish().to_pylist() == [
+        {
+            "kind": 4 if base_id is None else 5,
+            "data": None,
+            "uri": "_blobs/existing.blob",
+            "blob_id": base_id,
+            "blob_size": 0,
+            "position": 10,
+        }
+    ]
+    with pytest.raises(ValueError, match="already finished"):
+        builder.append_managed("_blobs/existing.blob", 0, 1)
+
+
+@pytest.mark.parametrize(
+    "uri,offset,size,message",
+    [
+        ("../outside.blob", 0, 1, "canonical relative object path"),
+        ("s3://bucket/object", 0, 1, "canonical relative object path"),
+        ("_blobs/existing.blob", 2**64 - 1, 1, "overflows u64"),
+    ],
+)
+def test_blob_descriptor_array_builder_rejects_invalid_managed(
+    uri, offset, size, message
+):
+    builder = lance.BlobDescriptorArrayBuilder("blob")
+    with pytest.raises(ValueError, match=message):
+        builder.append_managed(uri, offset, size)
+    assert len(builder.finish()) == 0
+
+
 def test_blob_descriptor_array_builder_writes_prepared_packed_blob_for_data_replacement(
     tmp_path,
 ):

python/src/blob.rs

@@ -237,6 +237,40 @@ impl PyBlobDescriptorArrayBuilder {
             .infer_error()
     }
 
+    /// Append a reference to an existing immutable, table-managed blob object.
+    ///
+    /// ``uri`` is relative to the destination table's base, or to the registered
+    /// ``base_id`` when supplied. ``offset`` and ``size`` select an exact byte
+    /// range; a zero size means an empty value. This does not upload bytes or
+    /// commit a reference. Keep the source snapshot protected until the new
+    /// reference commits, and use Managed-aware clients for table maintenance.
+    /// References shared across independent tables are not supported.
+    ///
+    /// Examples
+    /// --------
+    /// >>> builder = BlobDescriptorArrayBuilder("blob")
+    /// >>> builder.append_managed("_blobs/existing.blob", 0, 100)
+    /// >>> builder.append_managed("_blobs/existing.blob", 0, 100)
+    /// >>> len(builder.finish())
+    /// 2
+    #[pyo3(signature = (uri, offset, size, *, base_id=None))]
+    pub fn append_managed(
+        &mut self,
+        uri: String,
+        offset: u64,
+        size: u64,
+        base_id: Option<u32>,
+    ) -> PyResult<()> {
+        self.inner_mut()?
+            .push(BlobDescriptor::Managed {
+                base_id,
+                uri,
+                offset,
+                size,
+            })
+            .infer_error()
+    }
+
     pub fn append(&mut self, value: &Bound<'_, PyAny>) -> PyResult<()> {
         let value = value.extract::<pyo3::PyRef<'_, PyBlobDescriptor>>()?;
         self.inner_mut()?.push(value.inner.clone()).infer_error()

rust/lance-file/src/writer/field_types.rs

@@ -144,9 +144,15 @@ fn check_field(
         expected: expected.clone(),
         actual: actual.clone(),
     };
+    // Blob v2 has no serialized extension parameters. PyArrow serializes an
+    // empty string, while Rust's logical and prepared field builders omit it.
+    let expected_metadata = expected.extension_type_metadata();
+    let actual_metadata = actual.extension_type_metadata();
+    let metadata_matches = expected_metadata == actual_metadata
+        || (expected.extension_type_name() == Some(BLOB_V2_EXT_NAME)
+            && expected_metadata.unwrap_or_default() == actual_metadata.unwrap_or_default());
     if compare_extension
-        && (expected.extension_type_name() != actual.extension_type_name()
-            || expected.extension_type_metadata() != actual.extension_type_metadata())
+        && (expected.extension_type_name() != actual.extension_type_name() || !metadata_matches)
     {
         return Err(mismatch());
     }
@@ -191,3 +197,74 @@ fn check_field(
     }
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow_array::new_empty_array;
+    use arrow_schema::Schema as ArrowSchema;
+    use lance_arrow::{ARROW_EXT_META_KEY, ARROW_EXT_NAME_KEY};
+    use lance_core::datatypes::{BLOB_V2_LOGICAL_MINIMAL_FIELDS, BLOB_V2_PREPARED_TYPE};
+    use rstest::rstest;
+    use std::collections::HashMap;
+
+    #[rstest]
+    #[case::python_to_rust(BLOB_V2_EXT_NAME, BLOB_V2_EXT_NAME, Some(""), None, true)]
+    #[case::rust_to_python(BLOB_V2_EXT_NAME, BLOB_V2_EXT_NAME, None, Some(""), true)]
+    #[case::nonempty_expected(BLOB_V2_EXT_NAME, BLOB_V2_EXT_NAME, Some("parameters"), None, false)]
+    #[case::nonempty_actual(BLOB_V2_EXT_NAME, BLOB_V2_EXT_NAME, None, Some("parameters"), false)]
+    #[case::different_name(BLOB_V2_EXT_NAME, "other.extension", None, None, false)]
+    #[case::other_extension("other.extension", "other.extension", Some(""), None, false)]
+    fn blob_extension_metadata(
+        #[case] expected_name: &str,
+        #[case] actual_name: &str,
+        #[case] expected_metadata: Option<&str>,
+        #[case] actual_metadata: Option<&str>,
+        #[case] accepted: bool,
+    ) {
+        let nested_field = |data_type, name: &str, metadata: Option<&str>| {
+            let mut values = HashMap::from([(ARROW_EXT_NAME_KEY.to_string(), name.to_string())]);
+            if let Some(metadata) = metadata {
+                values.insert(ARROW_EXT_META_KEY.to_string(), metadata.to_string());
+            }
+            ArrowField::new(
+                "parent",
+                DataType::Struct(
+                    vec![ArrowField::new("blob", data_type, true).with_metadata(values)].into(),
+                ),
+                true,
+            )
+        };
+        let expected = nested_field(
+            DataType::Struct(BLOB_V2_LOGICAL_MINIMAL_FIELDS.clone()),
+            expected_name,
+            expected_metadata,
+        );
+        let actual_type = if expected_name == BLOB_V2_EXT_NAME {
+            BLOB_V2_PREPARED_TYPE.clone()
+        } else {
+            DataType::Struct(BLOB_V2_LOGICAL_MINIMAL_FIELDS.clone())
+        };
+        let actual = nested_field(actual_type, actual_name, actual_metadata);
+        let schema = Schema::try_from(&ArrowSchema::new(vec![expected])).unwrap();
+        let array = new_empty_array(actual.data_type());
+        let batch = RecordBatch::try_new(
+            Arc::new(ArrowSchema::new(vec![actual])),
+            vec![array.clone()],
+        )
+        .unwrap();
+        let mut expected_types = ExpectedTypes::new(&schema);
+        for result in [
+            expected_types.check_batch(&batch),
+            expected_types.check_column(0, &array),
+        ] {
+            if accepted {
+                result.unwrap();
+            } else {
+                let error = result.unwrap_err();
+                assert!(matches!(error, Error::InvalidInput { .. }));
+                assert!(error.to_string().contains("parent.blob"), "{error}");
+            }
+        }
+    }
+}