feat(python): accept expressions in update filters (#3876)

## Summary
- allow Python sync, async, and remote table updates to accept type-safe
`Expr` filters
- serialize expression filters before invoking the existing update
implementation
- cover numeric-looking text and apostrophe-containing text in sync and
async regression tests

## Root cause
`Table.update` was the remaining Python write path that required callers
to construct a raw SQL predicate. Dynamic text interpolated without SQL
literal encoding could therefore be parsed as an integer, float, or
unterminated string instead of Utf8. The expression API already encodes
literals safely for query and delete filters.

## Validation
- `cd python && .venv/bin/pytest
python/tests/test_table.py::test_update_async
python/tests/test_table.py::test_update_expr_filter_literals_async
python/tests/test_table.py::test_update
python/tests/test_table.py::test_update_expr_filter_literals -q`
- `cd python && .venv/bin/pytest python/tests/test_expr.py -q`
- `cd python && .venv/bin/ruff format --check .`
- `cd python && .venv/bin/ruff check .`

Fixes #1869

<!-- lance-gatekeeper-fix:v1 agent=01f1e7b69c65e8b6d3b3c1e1a7918179
generation=1 -->

---------

Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
Co-authored-by: Xuanwo <github@xuanwo.io>

python/python/tests/test_blob.py

@@ -8,7 +8,12 @@ import pyarrow.compute as pc
 import pytest
 
 import lancedb
-from lancedb._blob import read_row_ids_from_hits, stash_auto_row_ids
+from lancedb._blob import (
+    blob_v2_projection_sources,
+    read_row_ids_from_hits,
+    stash_auto_row_ids,
+)
+from lancedb.expr import col
 from lancedb.index import FTS
 from lancedb.schema import blob_column_paths, blob_v2_column_paths
 
@@ -70,6 +75,14 @@ def test_blob_v2_column_paths_include_list_children():
     ]
 
 
+def test_blob_v2_projection_sources_use_typed_column_name():
+    schema = pa.schema([lancedb.blob("blob")])
+
+    assert blob_v2_projection_sources(schema, {"blob_alias": col("blob")}) == {
+        "blob_alias": "blob"
+    }
+
+
 def _legacy_v1_table(name):
     db = lancedb.connect("memory:///")
     schema = pa.schema(
@@ -166,6 +179,20 @@ async def test_async_table_to_pandas_descriptions_mode_omits_row_id():
     assert set(descriptor.keys()) == {"kind", "position", "size", "blob_id", "blob_uri"}
 
 
+@pytest.mark.asyncio
+async def test_async_typed_blob_projection_preserves_source_column():
+    db = await lancedb.connect_async("memory:///typed_blob_projection")
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("blob")])
+    table = await db.create_table("typed_blob_projection", schema=schema)
+    await table.add([{"id": 1, "blob": b"alpha"}])
+
+    hits = await table.query().select({"blob_alias": col("blob")}).to_arrow()
+
+    assert "_lance_row_id" in hits.schema.field("blob_alias").type.names
+    blobs = await table.fetch_blobs("blob", hits)
+    assert blobs.to_pylist() == [b"alpha"]
+
+
 def test_fetch_blobs_round_trip():
     table = _blob_table(
         "round_trip",
@@ -403,6 +430,50 @@ async def test_blob_v2_hybrid_fetch_blobs_async():
     assert {blobs[i].as_py() for i in range(len(blobs))} == {b"alpha", b"beta"}
 
 
+@pytest.mark.asyncio
+async def test_async_hybrid_typed_blob_projection_preserves_source_column():
+    db = await lancedb.connect_async("memory:///hybrid_typed_blob")
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("text", pa.utf8()),
+            pa.field("vector", pa.list_(pa.float32(), list_size=2)),
+            lancedb.blob("blob"),
+        ]
+    )
+    table = await db.create_table("hybrid_typed_blob", schema=schema)
+    await table.add(
+        [
+            {
+                "id": 1,
+                "text": "hello alpha",
+                "vector": [1.0, 0.0],
+                "blob": b"alpha",
+            },
+            {
+                "id": 2,
+                "text": "hello beta",
+                "vector": [0.9, 0.1],
+                "blob": b"beta",
+            },
+        ]
+    )
+    await table.create_index("text", config=FTS(with_position=False))
+
+    hits = await (
+        table.query()
+        .nearest_to([1.0, 0.0])
+        .nearest_to_text("hello")
+        .select({"blob_alias": col("blob")})
+        .limit(2)
+        .to_arrow()
+    )
+
+    assert "_lance_row_id" in hits.schema.field("blob_alias").type.names
+    blobs = await table.fetch_blobs("blob", hits)
+    assert {blobs[i].as_py() for i in range(len(blobs))} == {b"alpha", b"beta"}
+
+
 def test_blob_file_seek_read_and_read_range():
     payload = _identifiable_payload(1024)
     table = _blob_table("seek_read", [{"id": 1, "image": payload}])