Merge remote-tracking branch 'origin/main' into gatekeeper/fix-2325-1

# Conflicts: # python/python/lancedb/remote/table.py
feat: support blob computed column refresh (#4081 )
2026-08-29 01:18:23 +00:00 · 2026-08-28 09:45:02 +00:00 · 2026-08-28 17:23:49 +08:00 · 2026-08-27 19:06:07 -07:00 · 2026-08-28 00:52:01 +00:00 · 2026-08-27 17:47:27 -07:00
55 changed files with 3114 additions and 603 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.38.0-beta.11"
+current_version = "0.38.0-beta.12"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -40,40 +40,31 @@ jobs:
          - target: aarch64-apple-darwin
            host: macos-latest
            features: fp16kernels
+            # Fat LTO was ~111 of this job's ~113 minutes.
+            lto: thin
+            codegen_units: 16
            pre_build: |-
              brew install protobuf
-              # Fat LTO (the workspace default in .cargo/config.toml) is
-              # single-threaded and is the peak-memory step of the build. On
-              # this runner it accounted for ~111 of the job's ~113 minutes,
-              # making it the critical path of the entire publish pipeline.
-              # ThinLTO parallelizes it across the runner's cores, for a few
-              # percent of runtime performance.
-              export CARGO_PROFILE_RELEASE_LTO=thin
-              export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
          - target: x86_64-pc-windows-msvc
            host: windows-2025
            features: ","
+            # The lower peak also keeps this on the standard 4-core runner.
+            lto: thin
+            codegen_units: 16
            pre_build: |-
              choco install --no-progress protoc ninja nasm
              tail -n 1000 /c/ProgramData/chocolatey/logs/chocolatey.log
              # There is an issue where choco doesn't add nasm to the path
              export PATH="$PATH:/c/Program Files/NASM"
              nasm -v
-              # See the ThinLTO note on aarch64-apple-darwin above. Keeping
-              # peak memory down is also what lets this run on the standard
-              # 4-core runner: the 8-core larger runner was only needed to
-              # stop fat LTO from OOMing rustc-LLVM.
-              export CARGO_PROFILE_RELEASE_LTO=thin
-              export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
          - target: aarch64-pc-windows-msvc
            host: windows-2025
            features: ","
+            lto: thin
+            codegen_units: 16
            pre_build: |-
                choco install --no-progress protoc
                rustup target add aarch64-pc-windows-msvc
-                # See the ThinLTO note on aarch64-apple-darwin above.
-                export CARGO_PROFILE_RELEASE_LTO=thin
-                export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
          - target: x86_64-unknown-linux-gnu
            host: ubuntu-latest
            features: fp16kernels
@@ -103,6 +94,14 @@ jobs:
            # https://github.com/napi-rs/napi-rs/blob/main/debian-aarch64.Dockerfile
            docker: ghcr.io/napi-rs/napi-rs/nodejs-rust:lts-debian-aarch64
            features: "fp16kernels"
+            # Fat LTO OOM-killed rustc every nightly; even with lld it peaked
+            # at 31391 MiB of the runner's 32 GiB.
+            lto: thin
+            codegen_units: 16
+            # arm64 Linux links through GNU `ld` where x86_64 defaults to
+            # `rust-lld`, which is why only arm64 OOM'd. lld cut the largest
+            # linker process 7.0 -> 4.0 GiB (lancedb/sophon#7313).
+            linker: /tmp/aarch64-lld-clang
            pre_build: |-
              set -e &&
              apt-get update &&
@@ -112,9 +111,30 @@ jobs:
              # AT_HWCAP2 (added in Linux 3.17). Define it for aws-lc-sys.
              export CFLAGS="$CFLAGS -DAT_HWCAP2=26" &&
              rustup target add aarch64-unknown-linux-gnu
+              # Not `&&`-chained: in dash, errexit does not fire for a
+              # non-final command in an `&&` list, so failures were ignored.
+              #
+              # A wrapper rather than `-C link-arg` because the per-target
+              # rustflags variable does not reach every unit that links, while
+              # the linker variable does. `clang` because GCC silently ignores
+              # `-fuse-ld=lld` unless built with lld support. Two echoes
+              # because printf's newline escape gets rewritten to `;` between
+              # here and the container.
+              echo '#!/bin/sh' > /tmp/aarch64-lld-clang
+              echo 'exec clang --target=aarch64-unknown-linux-gnu --sysroot=/usr/aarch64-unknown-linux-gnu/aarch64-unknown-linux-gnu/sysroot --gcc-toolchain=/usr/aarch64-unknown-linux-gnu -fuse-ld=lld "$@"' >> /tmp/aarch64-lld-clang
+              chmod 0755 /tmp/aarch64-lld-clang
+              # Fail now, not at the cdylib link ~30 minutes later. Linking at
+              # all also proves lld resolved; clang errors out when it cannot.
+              echo 'int main(void){return 0;}' > /tmp/probe.c
+              /tmp/aarch64-lld-clang /tmp/probe.c -o /tmp/probe
+              readelf -h /tmp/probe | grep AArch64
          - target: aarch64-unknown-linux-musl
            host: ubuntu-2404-8x-x64
            features: ","
+            # Fat LTO took the whole runner down. lld cannot help: it died
+            # inside rustc's LLVM, before any linker was spawned.
+            lto: thin
+            codegen_units: 16
            pre_build: |-
              set -e &&
              sudo apt-get update &&
@@ -123,6 +143,19 @@ jobs:
              export EXTRA_ARGS="-x"
    name: build - ${{ matrix.settings.target }}
    runs-on: ${{ matrix.settings.host }}
+    # On the job, not exported from `pre_build`: `Swatinem/rust-cache` hashes
+    # `CARGO_*` into its cache key before any step runs, so a step-local export
+    # leaves the key unchanged while cargo still rebuilds cold. The ThinLTO
+    # legs had been doing that every run.
+    #
+    # Not `RUSTFLAGS`: setting it, even to "", discards every config-file
+    # rustflag, silently dropping .cargo/config.toml's `target-cpu` and
+    # `target-feature` from the published binaries.
+    env:
+      CARGO_PROFILE_RELEASE_LTO: ${{ matrix.settings.lto || 'fat' }}
+      CARGO_PROFILE_RELEASE_CODEGEN_UNITS: ${{ matrix.settings.codegen_units || '1' }}
+      # Empty elsewhere: a per-target variable is only read for that triple.
+      CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER: ${{ matrix.settings.linker }}
    defaults:
      run:
        working-directory: nodejs
@@ -169,19 +202,15 @@ jobs:
          # creating ref). The nightly cadence also keeps entries inside
          # GitHub's 7-day eviction window, which a tag-only trigger would not.
          save-if: ${{ github.ref == 'refs/heads/main' }}
-      # Docker builds can use rust-cache too. `target/` already lives on the
-      # host because the whole workspace is bind-mounted into the container, and
-      # rust-cache's prune and save run host-side, so they can manage it -- which
-      # is what keeps the entry to dependency artifacts rather than a multi-GB
-      # copy of everything.
+      # Docker builds can use rust-cache too: the workspace is bind-mounted, so
+      # `target/` lives on the host and rust-cache's prune keeps the entry
+      # small.
      #
      # Two differences from the native builds. The container's CARGO_HOME is
-      # bind-mounted from `.cargo-cache` rather than the host's ~/.cargo, so that
-      # has to be cached explicitly. And the key is derived from the *host* rustc
-      # version, which is not the compiler that produced these artifacts; that is
-      # safe because cargo fingerprints the real compiler and rebuilds on a
-      # mismatch, it just means a base-image toolchain bump costs one cold build
-      # instead of invalidating the key.
+      # bind-mounted from `.cargo-cache` rather than ~/.cargo, so that is cached
+      # explicitly. And the key uses the *host* rustc version, not the compiler
+      # that built these artifacts -- safe, since cargo fingerprints the real
+      # one; a base-image bump just costs one cold build.
      - name: Cache cargo (docker builds)
        uses: Swatinem/rust-cache@v2
        if: ${{ matrix.settings.docker }}
@@ -210,9 +239,14 @@ jobs:
          # cache step above saves. Previously the registry mounts pointed at
          # `.cargo/...`, a path nothing cached, so the container re-downloaded
          # the whole crate registry on every run.
+          #
+          # `docker run` inherits nothing; `-e NAME` carries the job's `env:` in.
          options: "--user 0:0 -v ${{ github.workspace }}/.cargo-cache/git/db:/usr/local/cargo/git/db \
                    -v ${{ github.workspace }}/.cargo-cache/registry/cache:/usr/local/cargo/registry/cache \
                    -v ${{ github.workspace }}/.cargo-cache/registry/index:/usr/local/cargo/registry/index \
+                    -e CARGO_PROFILE_RELEASE_LTO \
+                    -e CARGO_PROFILE_RELEASE_CODEGEN_UNITS \
+                    -e CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER \
                    -v ${{ github.workspace }}:/build -w /build/nodejs"
          run: |
            set -e
@@ -256,6 +290,18 @@ jobs:
        if: always()
        run: df -h
        shell: bash
+      - name: Report peak memory
+        if: always() && runner.os == 'Linux'
+        shell: bash
+        run: |
+          peak=$(find /sys/fs/cgroup -name memory.peak -readable \
+                 -exec cat {} + 2>/dev/null | sort -n | tail -1)
+          if [ -n "$peak" ]; then
+            echo "peak memory: $((peak / 1024 / 1024)) MiB"
+          else
+            echo "peak memory: unavailable (no readable cgroup v2 memory.peak)"
+          fi
+          free -g || true
      - name: Upload artifact
        uses: actions/upload-artifact@v7
        with:
@@ -1597,9 +1597,9 @@ checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"

 [[package]]
 name = "chacha20"
-version = "0.10.0"
+version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601"
+checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
 dependencies = [
 "cfg-if 1.0.4",
 "cpufeatures 0.3.0",
@@ -5402,7 +5402,7 @@ dependencies = [

 [[package]]
 name = "lancedb"
-version = "0.38.0-beta.11"
+version = "0.38.0-beta.12"
 dependencies = [
 "ahash",
 "anyhow",
@@ -5490,7 +5490,7 @@ dependencies = [

 [[package]]
 name = "lancedb-nodejs"
-version = "0.38.0-beta.11"
+version = "0.38.0-beta.12"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5515,7 +5515,7 @@ dependencies = [

 [[package]]
 name = "lancedb-python"
-version = "0.38.0-beta.11"
+version = "0.38.0-beta.12"
 dependencies = [
 "arrow",
 "async-trait",
@@ -131,18 +131,13 @@ allow = [
    "BSD-3-Clause",
    "ISC",
    "Unicode-3.0",
-    "Unicode-DFS-2016",
    "Zlib",
    "CC0-1.0",
    "MPL-2.0",
    "BSL-1.0",
-    "OpenSSL",
    # 0BSD ("BSD Zero Clause") is effectively public domain — no attribution
    # required. Pulled in by `mock_instant`.
    "0BSD",
-    # bzip2-1.0.6 is the permissive upstream bzip2 license (BSD-like). Pulled
-    # in by `libbz2-rs-sys`, the pure-Rust bzip2 implementation.
-    "bzip2-1.0.6",
    # CDLA-Permissive-2.0 is a permissive data license used by `webpki-roots`
    # for the Mozilla CA root bundle. Data-only, distribution-compatible.
    "CDLA-Permissive-2.0",
@@ -150,12 +145,7 @@ allow = [
 confidence-threshold = 0.8
 # Per-crate license exceptions: allow a license for a specific crate only,
 # rather than globally via the `allow` list above.
-exceptions = [
-    # CDDL-1.0 (copyleft) is pulled in only as a dev/profiling dependency via
-    # `inferno` -> `pprof` -> `lance-testing`; it is a test dependency that we
-    # do not distribute, so scope the allowance to `inferno` alone.
-    { allow = ["CDDL-1.0"], crate = "inferno" },
-]
+exceptions = []
 # Crates whose license cannot be determined from Cargo metadata but whose
 # license we've manually confirmed from upstream. Keep this list minimal.
 [[licenses.clarify]]
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-core</artifactId>
-    <version>0.38.0-beta.11</version>
+    <version>0.38.0-beta.12</version>
 </dependency>
 ```

@@ -221,7 +221,7 @@ abstract checkpointLsm(): Promise<void>

 Converge this table's LSM write path into its base table.

-Freezes once, then triggers compaction and polls until the SSTables that existed
+Seals once, then triggers compaction and polls until the L0 that existed
 at the start is gone. The target set is fixed at the start, so
 generations created *during* the checkpoint are ignored — that is what
 lets it terminate under write load, and what makes it best-effort: it
@@ -289,7 +289,7 @@ It is a no-op when no writers are cached.
 abstract compactLsm(): Promise<void>
 ```

-Trigger a background SSTable compaction pass per table shard.
+Trigger a background L0 → base compaction pass per bucket.

 Returns once the passes are *dispatched*, not once they finish — watch
 [Table#getLsmStats](Table.md#getlsmstats) for progress, or use
@@ -505,7 +505,7 @@ Drop an index from the table.
 abstract flushLsm(): Promise<void>
 ```

-Freeze every table shard's active memtable into a new SSTable.
+Seal every bucket's active memtable into a new L0 generation.

 Returns once the seal is committed. Sealing an empty memtable is a no-op,
 so this is safe to call repeatedly.
@@ -519,10 +519,10 @@ so this is safe to call repeatedly.
 ### getLsmStats()

 ```ts
-abstract getLsmStats(includeSstableRows?): Promise<undefined | LsmStats>
+abstract getLsmStats(includeGenerationRows?): Promise<undefined | LsmStats>
 ```

-Read live per-table-shard LSM state.
+Read live per-bucket LSM state.

 Answers "how far behind is my fresh tier", "which bucket is hot", and
 "why is my fresh-tier vector search brute-force". Mutates no table state.
@@ -531,8 +531,8 @@ Resolves to `undefined` only when the LSM write path is not enabled.

 #### Parameters

-* **includeSstableRows?**: `boolean`
-    Also count rows per SSTable.
+* **includeGenerationRows?**: `boolean`
+    Also count rows per L0 generation.
    Off by default because each count opens an uncached Lance dataset.

 #### Returns
@@ -60,6 +60,7 @@
 - [BranchDiff](interfaces/BranchDiff.md)
 - [BranchIndexSummary](interfaces/BranchIndexSummary.md)
 - [BranchRowCountSummary](interfaces/BranchRowCountSummary.md)
+- [BucketStats](interfaces/BucketStats.md)
 - [CherryPickError](interfaces/CherryPickError.md)
 - [CherryPickPreview](interfaces/CherryPickPreview.md)
 - [CherryPickResult](interfaces/CherryPickResult.md)
@@ -86,6 +87,7 @@
 - [FtsToken](interfaces/FtsToken.md)
 - [FullTextQuery](interfaces/FullTextQuery.md)
 - [FullTextSearchOptions](interfaces/FullTextSearchOptions.md)
+- [GenerationStats](interfaces/GenerationStats.md)
 - [HnswPqOptions](interfaces/HnswPqOptions.md)
 - [HnswSqOptions](interfaces/HnswSqOptions.md)
 - [IndexConfig](interfaces/IndexConfig.md)
@@ -124,9 +126,7 @@
 - [SplitHashOptions](interfaces/SplitHashOptions.md)
 - [SplitRandomOptions](interfaces/SplitRandomOptions.md)
 - [SplitSequentialOptions](interfaces/SplitSequentialOptions.md)
- [SsTableStats](interfaces/SsTableStats.md)
 - [TableNamesOptions](interfaces/TableNamesOptions.md)
- [TableShardStats](interfaces/TableShardStats.md)
 - [TableStatistics](interfaces/TableStatistics.md)
 - [TimeoutConfig](interfaces/TimeoutConfig.md)
 - [TlsConfig](interfaces/TlsConfig.md)
@@ -2,12 +2,12 @@

 ***

-[@lancedb/lancedb](../globals.md) / TableShardStats
+[@lancedb/lancedb](../globals.md) / BucketStats

-# Interface: TableShardStats
+# Interface: BucketStats

-Live state of one table shard. A table is N table shards on one node; flattening to a
-single number hides the one hot table shard that is usually why someone opened
+Live state of one bucket. A table is N buckets on one node; flattening to a
+single number hides the one hot bucket that is usually why someone opened
 this endpoint.

 ## Properties
@@ -18,7 +18,7 @@ this endpoint.
 compacting: boolean;
 ```

-Whether a pass owns this table shard's compaction latch right now. Says *a*
+Whether a pass owns this bucket's compaction latch right now. Says *a*
 driver is running, not *whose*, and the latch is held from dispatch —
 including while the pass queues for a pod-wide compactor permit. Read it
 as "do not pile on", never as "mine is progressing".
@@ -35,13 +35,13 @@ The generation the active memtable will become.

 ***

-### sstables
+### generations

 ```ts
-sstables: SsTableStats[];
+generations: GenerationStats[];
 ```

-SSTables not yet merged into the base table.
+Flushed L0 generations not yet merged into the base table.

 ***

@@ -61,7 +61,7 @@ Version of the shard manifest these numbers were read from.
 optional memtables: MemtableStats[];
 ```

-Oldest first, active last. Absent for a `"Sealed"` table shard, whose
+Oldest first, active last. Absent for a `"Sealed"` bucket, whose
 in-memory state is torn down.

 ***
@@ -82,7 +82,7 @@ WAL position replay resumes from.
 shardId: string;
 ```

-The shard this table shard writes.
+The shard this bucket writes.

 ***

@@ -0,0 +1,40 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / GenerationStats
+
+# Interface: GenerationStats
+
+One flushed L0 generation.
+
+## Properties
+
+### bytes
+
+```ts
+bytes: number;
+```
+
+On-disk size of the generation.
+
+***
+
+### generation
+
+```ts
+generation: number;
+```
+
+The generation number. Increases as memtables are sealed into L0.
+
+***
+
+### rows?
+
+```ts
+optional rows: number;
+```
+
+Present only when `includeGenerationRows` was requested. Off by default
+because each count opens an uncached Lance dataset.
@@ -6,17 +6,17 @@

 # Interface: LsmStats

-Live per-table-shard LSM state, as returned by `Table#getLsmStats`.
+Live per-bucket LSM state, as returned by `Table#getLsmStats`.

-Nothing here is derived: sums and differences (total SSTable bytes, WAL lag) are
+Nothing here is derived: sums and differences (total L0 bytes, WAL lag) are
 the caller's to compute.

 ## Properties

-### tableShards
+### buckets

 ```ts
-tableShards: TableShardStats[];
+buckets: BucketStats[];
 ```

-One entry per table shard backing this table.
+One entry per bucket backing this table.
@@ -1,40 +0,0 @@
-[**@lancedb/lancedb**](../README.md) • **Docs**
-
-***
-
-[@lancedb/lancedb](../globals.md) / SsTableStats
-
-# Interface: SsTableStats
-
-One SSTable.
-
-## Properties
-
-### bytes
-
-```ts
-bytes: number;
-```
-
-On-disk size of the SSTable.
-
-***
-
-### generation
-
-```ts
-generation: number;
-```
-
-The generation number. Increases as memtables are frozen into SSTables.
-
-***
-
-### rows?
-
-```ts
-optional rows: number;
-```
-
-Present only when `includeSstableRows` was requested. Off by default
-because each count opens an uncached Lance dataset.
@@ -42,6 +42,8 @@ listing a storage directory.

 ::: lancedb.table.Table

+::: lancedb.table.CompactionOptions
+
 ::: lancedb.table.FragmentStatistics

 ::: lancedb.table.FragmentSummaryStats
@@ -223,9 +225,13 @@ tokens = list(
 Blob columns store large binary values out of line so they can be read lazily
 instead of being materialized with the rest of the row.

-::: lancedb.blob
+`lancedb.BlobType` is `lance.blob.BlobType` when pylance is installed. Without
+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.

-::: lancedb.BlobType
+::: lancedb.blob

 ::: lancedb._blob.BlobFile
    options:
@@ -8,7 +8,7 @@
    <parent>
      <groupId>com.lancedb</groupId>
      <artifactId>lancedb-parent</artifactId>
-      <version>0.38.0-beta.11</version>
+      <version>0.38.0-beta.12</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

@@ -22,11 +22,11 @@ import java.util.Optional;
 import java.util.OptionalLong;

 /**
- * Live state of one tableShard. A table is N tableShards on one node; flattening to a single number hides
- * the one hot tableShard that is usually why someone opened this endpoint.
+ * Live state of one bucket. A table is N buckets on one node; flattening to a single number hides
+ * the one hot bucket that is usually why someone opened this endpoint.
 */
-public class TableShardStats {
-  private static final String CONTEXT = "tableShard stats";
+public class BucketStats {
+  private static final String CONTEXT = "bucket stats";

  private final String shardId;
  private final String status;
@@ -35,11 +35,11 @@ public class TableShardStats {
  private final long currentGeneration;
  private final long replayAfterWalEntryPosition;
  private final long walEntryPositionLastSeen;
-  private final List<SsTableStats> sstables;
+  private final List<GenerationStats> generations;
  private final boolean compacting;
  private final List<MemtableStats> memtables;

-  TableShardStats(
+  BucketStats(
      String shardId,
      String status,
      long writerEpoch,
@@ -47,7 +47,7 @@ public class TableShardStats {
      long currentGeneration,
      long replayAfterWalEntryPosition,
      long walEntryPositionLastSeen,
-      List<SsTableStats> sstables,
+      List<GenerationStats> generations,
      boolean compacting,
      List<MemtableStats> memtables) {
    this.shardId = shardId;
@@ -57,12 +57,12 @@ public class TableShardStats {
    this.currentGeneration = currentGeneration;
    this.replayAfterWalEntryPosition = replayAfterWalEntryPosition;
    this.walEntryPositionLastSeen = walEntryPositionLastSeen;
-    this.sstables = Collections.unmodifiableList(sstables);
+    this.generations = Collections.unmodifiableList(generations);
    this.compacting = compacting;
    this.memtables = memtables == null ? null : Collections.unmodifiableList(memtables);
  }

-  /** The shard this tableShard writes. */
+  /** The shard this bucket writes. */
  public String shardId() {
    return shardId;
  }
@@ -100,13 +100,13 @@ public class TableShardStats {
    return walEntryPositionLastSeen;
  }

-  /** SSTables not yet merged into the base table. */
-  public List<SsTableStats> sstables() {
-    return sstables;
+  /** Flushed L0 generations not yet merged into the base table. */
+  public List<GenerationStats> generations() {
+    return generations;
  }

  /**
-   * Whether a pass owns this tableShard's compaction latch right now. Says <em>a</em> driver is
+   * Whether a pass owns this bucket's compaction latch right now. Says <em>a</em> driver is
   * running, not <em>whose</em>, and the latch is held from dispatch — including while the pass
   * queues for a pod-wide compactor permit. Read it as "do not pile on", never as "mine is
   * progressing".
@@ -115,15 +115,15 @@ public class TableShardStats {
    return compacting;
  }

-  /** Oldest first, active last. Empty for a {@code "Sealed"} tableShard, whose state is torn down. */
+  /** Oldest first, active last. Empty for a {@code "Sealed"} bucket, whose state is torn down. */
  public Optional<List<MemtableStats>> memtables() {
    return Optional.ofNullable(memtables);
  }

-  /** The newest SSTable generation, or empty when the tier is empty. */
-  OptionalLong newestSstableGeneration() {
+  /** The newest flushed generation, or empty when L0 is empty. */
+  OptionalLong newestGeneration() {
    OptionalLong newest = OptionalLong.empty();
-    for (SsTableStats generation : sstables) {
+    for (GenerationStats generation : generations) {
      if (!newest.isPresent() || generation.generation() > newest.getAsLong()) {
        newest = OptionalLong.of(generation.generation());
      }
@@ -132,15 +132,15 @@ public class TableShardStats {
  }

  /**
-   * How many SSTables at or below {@code target} are still uncompacted.
+   * How many generations at or below {@code target} are still in L0.
   *
   * <p>A count, not a boolean: one pass drains a bounded prefix rather than the whole target set,
   * so a boolean would read as "no progress" for every pass but the last. Compaction drains
   * oldest-first, so this decreases monotonically.
   */
-  long outstandingSstables(long target) {
+  long outstandingGenerations(long target) {
    long count = 0;
-    for (SsTableStats generation : sstables) {
+    for (GenerationStats generation : generations) {
      if (generation.generation() <= target) {
        count++;
      }
@@ -148,11 +148,11 @@ public class TableShardStats {
    return count;
  }

-  static TableShardStats fromJson(JsonNode node) {
+  static BucketStats fromJson(JsonNode node) {
    JsonFields.requiredObject(node, CONTEXT);
-    List<SsTableStats> sstables = new ArrayList<SsTableStats>();
-    for (JsonNode generation : JsonFields.requiredArray(node, "sstables", CONTEXT)) {
-      sstables.add(SsTableStats.fromJson(generation));
+    List<GenerationStats> generations = new ArrayList<GenerationStats>();
+    for (JsonNode generation : JsonFields.requiredArray(node, "generations", CONTEXT)) {
+      generations.add(GenerationStats.fromJson(generation));
    }

    JsonNode memtablesNode = JsonFields.optionalArray(node, "memtables", CONTEXT);
@@ -164,7 +164,7 @@ public class TableShardStats {
      }
    }

-    return new TableShardStats(
+    return new BucketStats(
        JsonFields.requiredText(node, "shard_id", CONTEXT),
        JsonFields.requiredText(node, "status", CONTEXT),
        JsonFields.requiredLong(node, "writer_epoch", CONTEXT),
@@ -172,21 +172,21 @@ public class TableShardStats {
        JsonFields.requiredLong(node, "current_generation", CONTEXT),
        JsonFields.requiredLong(node, "replay_after_wal_entry_position", CONTEXT),
        JsonFields.requiredLong(node, "wal_entry_position_last_seen", CONTEXT),
-        sstables,
+        generations,
        JsonFields.requiredBoolean(node, "compacting", CONTEXT),
        memtables);
  }

  @Override
  public String toString() {
-    return "TableShardStats{shardId="
+    return "BucketStats{shardId="
        + shardId
        + ", status="
        + status
        + ", currentGeneration="
        + currentGeneration
-        + ", sstables="
-        + sstables
+        + ", generations="
+        + generations
        + ", compacting="
        + compacting
        + "}";
@@ -17,21 +17,21 @@ import com.fasterxml.jackson.databind.JsonNode;

 import java.util.OptionalLong;

-/** One SSTable. */
-public class SsTableStats {
+/** One flushed L0 generation. */
+public class GenerationStats {
  private static final String CONTEXT = "generation stats";

  private final long generation;
  private final long bytes;
  private final Long rows;

-  SsTableStats(long generation, long bytes, Long rows) {
+  GenerationStats(long generation, long bytes, Long rows) {
    this.generation = generation;
    this.bytes = bytes;
    this.rows = rows;
  }

-  /** The generation number. Increases as memtables are frozen into SSTables. */
+  /** The generation number. Increases as memtables are sealed into L0. */
  public long generation() {
    return generation;
  }
@@ -42,16 +42,16 @@ public class SsTableStats {
  }

  /**
-   * Rows in this generation, present only when {@code includeSstableRows} was requested. Off by
+   * Rows in this generation, present only when {@code includeGenerationRows} was requested. Off by
   * default because each count opens an uncached Lance dataset.
   */
  public OptionalLong rows() {
    return rows == null ? OptionalLong.empty() : OptionalLong.of(rows);
  }

-  static SsTableStats fromJson(JsonNode node) {
+  static GenerationStats fromJson(JsonNode node) {
    JsonFields.requiredObject(node, CONTEXT);
-    return new SsTableStats(
+    return new GenerationStats(
        JsonFields.requiredLong(node, "generation", CONTEXT),
        JsonFields.requiredLong(node, "bytes", CONTEXT),
        JsonFields.optionalLong(node, "rows", CONTEXT));
@@ -59,6 +59,6 @@ public class SsTableStats {

  @Override
  public String toString() {
-    return "SsTableStats{generation=" + generation + ", bytes=" + bytes + ", rows=" + rows + "}";
+    return "GenerationStats{generation=" + generation + ", bytes=" + bytes + ", rows=" + rows + "}";
  }
 }
@@ -26,7 +26,7 @@ import java.util.OptionalLong;
 *
 * <p>Installing an {@link LsmWriteSpec} routes {@code mergeInsert} upserts through Lance's MemWAL —
 * an LSM-style append — instead of the standard merge path. Rows land in an in-memory memtable,
- * freeze into SSTables, and are merged into the base table by compaction.
+ * seal into L0 generations, and are merged into the base table by compaction.
 *
 * <p>These routes are not part of the Lance Namespace specification, so they are issued directly
 * rather than through {@link org.lance.namespace.LanceNamespace}.
@@ -38,7 +38,7 @@ import java.util.OptionalLong;
 *     .buildRestClient();
 *
 * LanceDbTableLsm lsm = new LanceDbTableLsm(client, "my_table");
- * lsm.setLsmWriteSpec(LsmWriteSpec.tableShard("id", 16));
+ * lsm.setLsmWriteSpec(LsmWriteSpec.bucket("id", 16));
 * // ... merge_insert traffic ...
 * lsm.checkpointLsm();
 * }</pre>
@@ -94,7 +94,7 @@ public class LanceDbTableLsm {
   * Install an {@link LsmWriteSpec} on this table, selecting the MemWAL LSM write path for future
   * {@code mergeInsert} calls.
   *
-   * <p>All variants require the table to have an unenforced primary key; tableShard sharding
+   * <p>All variants require the table to have an unenforced primary key; bucket sharding
   * additionally requires it to be the single column being bucketed.
   */
  public void setLsmWriteSpec(LsmWriteSpec spec) {
@@ -130,7 +130,7 @@ public class LanceDbTableLsm {
  }

  /**
-   * Freeze every table shard's active memtable into a new SSTable.
+   * Seal every bucket's active memtable into a new L0 generation.
   *
   * <p>Returns once the seal is committed. Sealing an empty memtable is a no-op, so this is safe to
   * call repeatedly.
@@ -140,7 +140,7 @@ public class LanceDbTableLsm {
  }

  /**
-   * Trigger a background SSTable compaction pass per table shard.
+   * Trigger a background L0 → base compaction pass per bucket.
   *
   * <p>Returns once the passes are <em>dispatched</em>, not once they finish — watch {@link
   * #getLsmStats}, or use {@link #checkpointLsm} to wait for convergence.
@@ -150,9 +150,9 @@ public class LanceDbTableLsm {
  }

  /**
-   * Read live per-tableShard LSM state.
+   * Read live per-bucket LSM state.
   *
-   * <p>Answers "how far behind is my fresh tier", "which tableShard is hot", and "why is my fresh-tier
+   * <p>Answers "how far behind is my fresh tier", "which bucket is hot", and "why is my fresh-tier
   * vector search brute-force". Mutates no table state.
   *
   * <p>Empty only when the LSM write path is not enabled — that is, when the server sends an absent
@@ -160,13 +160,13 @@ public class LanceDbTableLsm {
   * one throws rather than decoding to something empty, because {@link #checkpointLsm} reads
   * convergence out of these numbers and cannot tell a defaulted array from a drained one.
   *
-   * @param includeSstableRows Also count rows per SSTable. Off by default because each
+   * @param includeGenerationRows Also count rows per L0 generation. Off by default because each
   *     count opens an uncached Lance dataset.
   * @throws IllegalStateException if the response is absent or does not decode.
   */
-  public Optional<LsmStats> getLsmStats(boolean includeSstableRows) {
+  public Optional<LsmStats> getLsmStats(boolean includeGenerationRows) {
    Map<String, Object> body = new LinkedHashMap<String, Object>();
-    body.put("include_sstable_rows", includeSstableRows);
+    body.put("include_generation_rows", includeGenerationRows);
    JsonNode response = client.post(route("get_lsm_stats"), body);
    if (response == null) {
      throw new IllegalStateException("get_lsm_stats returned an empty response body");
@@ -186,8 +186,8 @@ public class LanceDbTableLsm {
  /**
   * Converge this table's LSM write path into its base table.
   *
-   * <p>Freezes once, fixes a target watermark from the resulting SSTables, then triggers compaction and
-   * polls until those SSTables are gone. The target set is fixed at the start, so sstables created
+   * <p>Seals once, fixes a target watermark from the resulting L0, then triggers compaction and
+   * polls until that L0 is gone. The target set is fixed at the start, so generations created
   * <em>during</em> the checkpoint are ignored — that is what lets it terminate under write load,
   * and what makes it best-effort: it converges the fresh tier as of some instant. Idempotent,
   * abandonable at any point, safe on a cadence.
@@ -204,7 +204,7 @@ public class LanceDbTableLsm {
    for (int reissue = 0; reissue <= MAX_REISSUES; reissue++) {
      // The seal turns everything written before this call into a generation, so the
      // watermark has to be read after it. Idempotent: sealing an empty memtable is a
-      // no-op, so a re-issue does not churn empty sstables.
+      // no-op, so a re-issue does not churn empty generations.
      if (issueVoid(this::flushLsm)) {
        backoff(reissue);
        continue;
@@ -220,7 +220,7 @@ public class LanceDbTableLsm {
        return;
      }

-      Map<String, Long> targets = newestSstableGenerations(stats.value.get());
+      Map<String, Long> targets = newestGenerations(stats.value.get());
      if (targets.isEmpty()) {
        return;
      }
@@ -236,7 +236,7 @@ public class LanceDbTableLsm {
  }

  /**
-   * Trigger and poll until no tableShard holds a generation at or below its target.
+   * Trigger and poll until no bucket holds a generation at or below its target.
   *
   * @return true when the drain finished, false when the table needs re-claiming from flush.
   */
@@ -250,21 +250,21 @@ public class LanceDbTableLsm {
        return true;
      }

-      // `compacting` is the tableShard's compaction latch, held from dispatch until the pass
+      // `compacting` is the bucket's compaction latch, held from dispatch until the pass
      // ends — including while it waits on a pod-wide permit. So it answers one question
      // only: do not pile on. Buckets with nothing outstanding are skipped, not counted
      // as idle.
      long outstanding = 0;
      boolean allCompacting = true;
-      for (TableShardStats tableShard : stats.value.get().tableShards()) {
-        Long target = targets.get(tableShard.shardId());
+      for (BucketStats bucket : stats.value.get().buckets()) {
+        Long target = targets.get(bucket.shardId());
        if (target == null) {
          continue;
        }
-        long remaining = tableShard.outstandingSstables(target);
+        long remaining = bucket.outstandingGenerations(target);
        if (remaining > 0) {
          outstanding += remaining;
-          allCompacting &= tableShard.compacting();
+          allCompacting &= bucket.compacting();
        }
      }
      if (outstanding == 0) {
@@ -281,7 +281,7 @@ public class LanceDbTableLsm {
          if (!isRetryable(e)) {
            throw e;
          }
-          // A 429 here means the server could latch no tableShard at all, which the poll
+          // A 429 here means the server could latch no bucket at all, which the poll
          // above already handles. Not retried in place: the latch it would contend for
          // is the one doing the work, so fall through and re-read — POLL_INTERVAL_MS is
          // the backoff.
@@ -291,13 +291,13 @@ public class LanceDbTableLsm {
    }
  }

-  /** The newest generation held by each tableShard, skipping tableShards holding none. */
-  private static Map<String, Long> newestSstableGenerations(LsmStats stats) {
+  /** The newest generation held by each bucket, skipping buckets holding none. */
+  private static Map<String, Long> newestGenerations(LsmStats stats) {
    Map<String, Long> targets = new HashMap<String, Long>();
-    for (TableShardStats tableShard : stats.tableShards()) {
-      OptionalLong newest = tableShard.newestSstableGeneration();
+    for (BucketStats bucket : stats.buckets()) {
+      OptionalLong newest = bucket.newestGeneration();
      if (newest.isPresent()) {
-        targets.put(tableShard.shardId(), newest.getAsLong());
+        targets.put(bucket.shardId(), newest.getAsLong());
      }
    }
    return targets;
@@ -20,37 +20,37 @@ import java.util.Collections;
 import java.util.List;

 /**
- * Live per-tableShard LSM state, as returned by {@link LanceDbTableLsm#getLsmStats()}.
+ * Live per-bucket LSM state, as returned by {@link LanceDbTableLsm#getLsmStats()}.
 *
- * <p>Nothing here is derived: sums and differences (total SSTable bytes, WAL lag) are the caller's to
+ * <p>Nothing here is derived: sums and differences (total L0 bytes, WAL lag) are the caller's to
 * compute. There is no "LSM is off" shape — that case is an empty {@link java.util.Optional},
 * because a stats object of zeros would read as measurements.
 */
 public class LsmStats {
  private static final String CONTEXT = "lsm stats";

-  private final List<TableShardStats> tableShards;
+  private final List<BucketStats> buckets;

-  LsmStats(List<TableShardStats> tableShards) {
-    this.tableShards = Collections.unmodifiableList(tableShards);
+  LsmStats(List<BucketStats> buckets) {
+    this.buckets = Collections.unmodifiableList(buckets);
  }

-  /** One entry per tableShard. */
-  public List<TableShardStats> tableShards() {
-    return tableShards;
+  /** One entry per bucket. */
+  public List<BucketStats> buckets() {
+    return buckets;
  }

  static LsmStats fromJson(JsonNode node) {
    JsonFields.requiredObject(node, CONTEXT);
-    List<TableShardStats> tableShards = new ArrayList<TableShardStats>();
-    for (JsonNode tableShard : JsonFields.requiredArray(node, "table_shards", CONTEXT)) {
-      tableShards.add(TableShardStats.fromJson(tableShard));
+    List<BucketStats> buckets = new ArrayList<BucketStats>();
+    for (JsonNode bucket : JsonFields.requiredArray(node, "buckets", CONTEXT)) {
+      buckets.add(BucketStats.fromJson(bucket));
    }
-    return new LsmStats(tableShards);
+    return new LsmStats(buckets);
  }

  @Override
  public String toString() {
-    return "LsmStats{tableShards=" + tableShards + "}";
+    return "LsmStats{buckets=" + buckets + "}";
  }
 }
@@ -132,10 +132,10 @@ public class LanceDbTableLsmTest {
    enqueue("set_lsm_write_spec", 200, "");

    lsm.setLsmWriteSpec(
-        LsmWriteSpec.tableShard("id", 16).withMaintainedIndexes(Arrays.asList("id_idx")));
+        LsmWriteSpec.bucket("id", 16).withMaintainedIndexes(Arrays.asList("id_idx")));

    JsonNode body = MAPPER.readTree(requestBodies.get(0));
-    assertEquals("tableShard", body.get("sharding").get("mode").asText());
+    assertEquals("bucket", body.get("sharding").get("mode").asText());
    assertEquals("id", body.get("sharding").get("column").asText());
    assertEquals(16, body.get("sharding").get("num_buckets").asInt());
    assertEquals(1, body.get("maintained_indexes").size());
@@ -201,7 +201,7 @@ public class LanceDbTableLsmTest {
    enqueue(
        "get_lsm_write_spec",
        200,
-        "{\"lsm_write_spec\":{\"sharding\":{\"mode\":\"tableShard\",\"column\":\"id\","
+        "{\"lsm_write_spec\":{\"sharding\":{\"mode\":\"bucket\",\"column\":\"id\","
            + "\"num_buckets\":16},\"maintained_indexes\":[\"id_idx\"],"
            + "\"writer_config_defaults\":{\"durable_write\":\"true\"}}}");

@@ -228,14 +228,14 @@ public class LanceDbTableLsmTest {

  @Test
  public void testGetLsmStats() throws Exception {
-    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));

    Optional<LsmStats> got = lsm.getLsmStats(true);

    assertEquals("/v1/table/my_table/get_lsm_stats/", requestPaths.get(0));
-    assertTrue(MAPPER.readTree(requestBodies.get(0)).get("include_sstable_rows").asBoolean());
+    assertTrue(MAPPER.readTree(requestBodies.get(0)).get("include_generation_rows").asBoolean());
    assertTrue(got.isPresent());
-    TableShardStats decoded = got.get().tableShards().get(0);
+    BucketStats decoded = got.get().buckets().get(0);
    assertEquals("shard-0", decoded.shardId());
    assertEquals("Active", decoded.status());
    assertEquals(1, decoded.writerEpoch());
@@ -243,8 +243,8 @@ public class LanceDbTableLsmTest {
    assertEquals(9, decoded.currentGeneration());
    assertFalse(decoded.compacting());
    assertEquals(Arrays.asList(7L, 8L), generationNumbers(decoded));
-    assertEquals(1024, decoded.sstables().get(0).bytes());
-    assertFalse(decoded.sstables().get(0).rows().isPresent(), "rows absent unless requested");
+    assertEquals(1024, decoded.generations().get(0).bytes());
+    assertFalse(decoded.generations().get(0).rows().isPresent(), "rows absent unless requested");
    assertFalse(decoded.memtables().isPresent(), "absent memtables stay absent");
  }

@@ -254,19 +254,19 @@ public class LanceDbTableLsmTest {
    enqueue(
        "get_lsm_stats",
        200,
-        "{\"lsm_stats\":{\"tableShards\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+        "{\"lsm_stats\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
            + "\"writer_epoch\":1,\"manifest_version\":2,\"current_generation\":9,"
            + "\"replay_after_wal_entry_position\":3,\"wal_entry_position_last_seen\":11,"
-            + "\"sstables\":[{\"generation\":7,\"bytes\":1024,\"rows\":42}],"
+            + "\"generations\":[{\"generation\":7,\"bytes\":1024,\"rows\":42}],"
            + "\"compacting\":true,\"memtables\":[{\"generation\":8,\"rows\":5,"
            + "\"bytes\":64,\"batches\":2,\"indexes\":[\"id_idx\"]}]}]}}");

-    TableShardStats decoded = lsm.getLsmStats(true).get().tableShards().get(0);
+    BucketStats decoded = lsm.getLsmStats(true).get().buckets().get(0);

    assertEquals(3, decoded.replayAfterWalEntryPosition());
    assertEquals(11, decoded.walEntryPositionLastSeen());
    assertTrue(decoded.compacting());
-    assertEquals(42, decoded.sstables().get(0).rows().getAsLong());
+    assertEquals(42, decoded.generations().get(0).rows().getAsLong());
    assertTrue(decoded.memtables().isPresent());
    MemtableStats memtable = decoded.memtables().get().get(0);
    assertEquals(8, memtable.generation());
@@ -289,7 +289,7 @@ public class LanceDbTableLsmTest {

    lsm.getLsmStats();

-    assertFalse(MAPPER.readTree(requestBodies.get(0)).get("include_sstable_rows").asBoolean());
+    assertFalse(MAPPER.readTree(requestBodies.get(0)).get("include_generation_rows").asBoolean());
  }

  // ===========================================================================
@@ -334,8 +334,8 @@ public class LanceDbTableLsmTest {
  @Test
  public void testCheckpointReturnsWhenNoGenerationsOutstanding() {
    enqueue("flush_lsm", 200, "");
-    // A table shard with no SSTables yields no target, so the drain never starts.
-    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false)));
+    // A bucket with no L0 generations yields no target, so the drain never starts.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false)));

    lsm.checkpointLsm();

@@ -345,12 +345,12 @@ public class LanceDbTableLsmTest {
  @Test
  public void testCheckpointConvergesOnceTargetGenerationsAreGone() {
    enqueue("flush_lsm", 200, "");
-    // Watermark read: shard-0 holds sstables 7 and 8, so target = 8.
-    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
+    // Watermark read: shard-0 holds generations 7 and 8, so target = 8.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
    // First drain poll: both still outstanding, nothing compacting -> dispatch a pass.
-    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
    // Second drain poll: drained past the target -> done.
-    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false, 9L)));
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 9L)));
    enqueue("compact_lsm", 200, "");

    lsm.checkpointLsm();
@@ -362,14 +362,14 @@ public class LanceDbTableLsmTest {
  @Test
  public void testCheckpointDoesNotPileOnWhileEveryTargetBucketIsCompacting() {
    enqueue("flush_lsm", 200, "");
-    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", true, 4L)));
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", true, 4L)));
    // Still compacting on the first poll, so no pass is dispatched; then it drains.
-    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", true, 4L)));
-    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false, 5L)));
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", true, 4L)));
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 5L)));

    lsm.checkpointLsm();

-    assertEquals(0, countCalls("compact_lsm"), "a latched tableShard is left alone");
+    assertEquals(0, countCalls("compact_lsm"), "a latched bucket is left alone");
  }

  @Test
@@ -378,7 +378,7 @@ public class LanceDbTableLsmTest {
    // from flush rather than retrying the read in place.
    enqueue("flush_lsm", 200, "");
    enqueue("get_lsm_stats", 421, "no claim");
-    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false)));
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false)));

    lsm.checkpointLsm();

@@ -389,7 +389,7 @@ public class LanceDbTableLsmTest {
  public void testCheckpointRetriesRetryableStatusInPlace() {
    enqueue("flush_lsm", 429, "latch held");
    enqueue("flush_lsm", 200, "");
-    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false)));
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false)));

    lsm.checkpointLsm();

@@ -421,27 +421,27 @@ public class LanceDbTableLsmTest {

  /**
   * A stats payload that does not decode must fail closed. Every one of these bodies used to be
-   * read as "no tableShards", which is indistinguishable from a drained table, so {@code checkpointLsm}
+   * read as "no buckets", which is indistinguishable from a drained table, so {@code checkpointLsm}
   * reported convergence for a checkpoint that never ran.
   */
  @Test
  public void testCheckpointRejectsMalformedStats() {
    Map<String, String> malformed = new LinkedHashMap<String, String>();
    malformed.put("no response body at all", "");
-    malformed.put("stats object with no tableShards", "{\"lsm_stats\":{}}");
-    malformed.put("tableShard missing its required fields", "{\"lsm_stats\":{\"tableShards\":[{}]}}");
+    malformed.put("stats object with no buckets", "{\"lsm_stats\":{}}");
+    malformed.put("bucket missing its required fields", "{\"lsm_stats\":{\"buckets\":[{}]}}");
    malformed.put(
-        "tableShard missing sstables",
-        "{\"lsm_stats\":{\"tableShards\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+        "bucket missing generations",
+        "{\"lsm_stats\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
            + "\"writer_epoch\":1,\"manifest_version\":2,\"current_generation\":9,"
            + "\"replay_after_wal_entry_position\":0,\"wal_entry_position_last_seen\":0,"
            + "\"compacting\":false}]}}");
    malformed.put(
        "generation with a non-numeric generation number",
-        "{\"lsm_stats\":{\"tableShards\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+        "{\"lsm_stats\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
            + "\"writer_epoch\":1,\"manifest_version\":2,\"current_generation\":9,"
            + "\"replay_after_wal_entry_position\":0,\"wal_entry_position_last_seen\":0,"
-            + "\"sstables\":[{\"generation\":\"7\",\"bytes\":1024}],"
+            + "\"generations\":[{\"generation\":\"7\",\"bytes\":1024}],"
            + "\"compacting\":false}]}}");

    for (Map.Entry<String, String> each : malformed.entrySet()) {
@@ -492,22 +492,22 @@ public class LanceDbTableLsmTest {
  // harness
  // ===========================================================================

-  private static List<Long> generationNumbers(TableShardStats tableShard) {
+  private static List<Long> generationNumbers(BucketStats bucket) {
    List<Long> numbers = new ArrayList<Long>();
-    for (SsTableStats generation : tableShard.sstables()) {
+    for (GenerationStats generation : bucket.generations()) {
      numbers.add(generation.generation());
    }
    return numbers;
  }

-  /** Build an {@code lsm_stats} response body from tableShard fragments. */
-  private static String stats(String... tableShards) {
-    return "{\"lsm_stats\":{\"tableShards\":[" + String.join(",", tableShards) + "]}}";
+  /** Build an {@code lsm_stats} response body from bucket fragments. */
+  private static String stats(String... buckets) {
+    return "{\"lsm_stats\":{\"buckets\":[" + String.join(",", buckets) + "]}}";
  }

-  private static String tableShard(String shardId, boolean compacting, Long... sstables) {
+  private static String bucket(String shardId, boolean compacting, Long... generations) {
    StringBuilder gens = new StringBuilder();
-    for (Long generation : sstables) {
+    for (Long generation : generations) {
      if (gens.length() > 0) {
        gens.append(",");
      }
@@ -517,7 +517,7 @@ public class LanceDbTableLsmTest {
        + shardId
        + "\",\"status\":\"Active\",\"writer_epoch\":1,\"manifest_version\":2,"
        + "\"current_generation\":9,\"replay_after_wal_entry_position\":0,"
-        + "\"wal_entry_position_last_seen\":0,\"sstables\":["
+        + "\"wal_entry_position_last_seen\":0,\"generations\":["
        + gens
        + "],\"compacting\":"
        + compacting
@@ -6,7 +6,7 @@

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.38.0-beta.11</version>
+    <version>0.38.0-beta.12</version>
    <packaging>pom</packaging>
    <name>${project.artifactId}</name>
    <description>LanceDB Java SDK Parent POM</description>
@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.38.0-beta.11"
+version = "0.38.0-beta.12"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -157,8 +157,8 @@ export {
  TokenizeTableOptions,
  LsmWriteSpec,
  LsmStats,
-  TableShardStats,
-  SsTableStats,
+  BucketStats,
+  GenerationStats,
  MemtableStats,
  ColumnAlteration,
  FieldMetadataUpdate,
@@ -55,8 +55,8 @@ import { sanitizeType } from "./sanitize";
 import { IntoSql, toSQL } from "./util";
 export { IndexConfig } from "./native";
 export {
-  TableShardStats,
-  SsTableStats,
+  BucketStats,
+  GenerationStats,
  LsmStats,
  MemtableStats,
 } from "./native";
@@ -741,7 +741,7 @@ export abstract class Table {
   */
  abstract closeLsmWriters(): Promise<void>;
  /**
-   * Freeze every table shard's active memtable into a new SSTable.
+   * Seal every bucket's active memtable into a new L0 generation.
   *
   * Returns once the seal is committed. Sealing an empty memtable is a no-op,
   * so this is safe to call repeatedly.
@@ -749,7 +749,7 @@ export abstract class Table {
   */
  abstract flushLsm(): Promise<void>;
  /**
-   * Trigger a background SSTable compaction pass per table shard.
+   * Trigger a background L0 → base compaction pass per bucket.
   *
   * Returns once the passes are *dispatched*, not once they finish — watch
   * {@link Table#getLsmStats} for progress, or use
@@ -760,9 +760,9 @@ export abstract class Table {
  /**
   * Converge this table's LSM write path into its base table.
   *
-   * Freezes once, then triggers compaction and polls until the SSTables that existed
+   * Seals once, then triggers compaction and polls until the L0 that existed
   * at the start is gone. The target set is fixed at the start, so
-   * SSTables created *during* the checkpoint are ignored — that is what
+   * generations created *during* the checkpoint are ignored — that is what
   * lets it terminate under write load, and what makes it best-effort: it
   * converges the fresh tier as of some instant. Idempotent, abandonable at
   * any point, and safe to run on a cadence.
@@ -786,12 +786,12 @@ export abstract class Table {
   * "why is my fresh-tier vector search brute-force". Mutates no table state.
   *
   * Resolves to `undefined` only when the LSM write path is not enabled.
-   * @param {boolean} includeSstableRows Also count rows per SSTable.
+   * @param {boolean} includeGenerationRows Also count rows per L0 generation.
   *   Off by default because each count opens an uncached Lance dataset.
   * @returns {Promise<LsmStats | undefined>}
   */
  abstract getLsmStats(
-    includeSstableRows?: boolean,
+    includeGenerationRows?: boolean,
  ): Promise<LsmStats | undefined>;
  /** Retrieve the version of the table */

@@ -1388,9 +1388,9 @@ export class LocalTable extends Table {
  }

  async getLsmStats(
-    includeSstableRows: boolean = false,
+    includeGenerationRows: boolean = false,
  ): Promise<LsmStats | undefined> {
-    return (await this.inner.getLsmStats(includeSstableRows)) ?? undefined;
+    return (await this.inner.getLsmStats(includeGenerationRows)) ?? undefined;
  }

  async version(): Promise<number> {
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.38.0-beta.11",
+	"version": "0.38.0-beta.12",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.38.0-beta.11",
+	"version": "0.38.0-beta.12",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.38.0-beta.11",
+	"version": "0.38.0-beta.12",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.38.0-beta.11",
+	"version": "0.38.0-beta.12",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.38.0-beta.11",
+	"version": "0.38.0-beta.12",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.38.0-beta.11",
+  "version": "0.38.0-beta.12",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.38.0-beta.11",
+	"version": "0.38.0-beta.12",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -1,12 +1,12 @@
 {
  "name": "@lancedb/lancedb",
-  "version": "0.38.0-beta.11",
+  "version": "0.38.0-beta.12",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "@lancedb/lancedb",
-      "version": "0.38.0-beta.11",
+      "version": "0.38.0-beta.12",
      "cpu": [
        "x64",
        "arm64"
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.38.0-beta.11",
+  "version": "0.38.0-beta.12",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -542,11 +542,11 @@ impl Table {
    #[napi(catch_unwind)]
    pub async fn get_lsm_stats(
        &self,
-        include_sstable_rows: bool,
+        include_generation_rows: bool,
    ) -> napi::Result<Option<LsmStats>> {
        let stats = self
            .inner_ref()?
-            .get_lsm_stats(include_sstable_rows)
+            .get_lsm_stats(include_generation_rows)
            .await
            .default_error()?;
        Ok(stats.map(LsmStats::from))
@@ -950,21 +950,21 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
    }
 }

-/// One SSTable.
+/// One flushed L0 generation.
 #[napi(object)]
 #[derive(Clone, Debug)]
-pub struct SsTableStats {
-    /// The generation number. Increases as memtables are frozen into SSTables.
+pub struct GenerationStats {
+    /// The generation number. Increases as memtables are sealed into L0.
    pub generation: i64,
-    /// On-disk size of the SSTable.
+    /// On-disk size of the generation.
    pub bytes: i64,
-    /// Present only when `includeSstableRows` was requested. Off by default
+    /// Present only when `includeGenerationRows` was requested. Off by default
    /// because each count opens an uncached Lance dataset.
    pub rows: Option<i64>,
 }

-impl From<lancedb::table::SsTableStats> for SsTableStats {
-    fn from(g: lancedb::table::SsTableStats) -> Self {
+impl From<lancedb::table::GenerationStats> for GenerationStats {
+    fn from(g: lancedb::table::GenerationStats) -> Self {
        Self {
            generation: g.generation as i64,
            bytes: g.bytes as i64,
@@ -977,7 +977,7 @@ impl From<lancedb::table::SsTableStats> for SsTableStats {
 #[napi(object)]
 #[derive(Clone, Debug)]
 pub struct MemtableStats {
-    /// The generation this memtable will become once frozen.
+    /// The generation this memtable will become once sealed.
    pub generation: i64,
    /// Rows currently buffered.
    pub rows: i64,
@@ -1002,13 +1002,13 @@ impl From<lancedb::table::MemtableStats> for MemtableStats {
    }
 }

-/// Live state of one table shard. A table is N table shards on one node; flattening to a
-/// single number hides the one hot table shard that is usually why someone opened
+/// Live state of one bucket. A table is N buckets on one node; flattening to a
+/// single number hides the one hot bucket that is usually why someone opened
 /// this endpoint.
 #[napi(object)]
 #[derive(Clone, Debug)]
-pub struct TableShardStats {
-    /// The shard this table shard writes.
+pub struct BucketStats {
+    /// The shard this bucket writes.
    pub shard_id: String,
    /// `"Active"` or `"Sealed"` (drop-table 2PC in flight).
    pub status: String,
@@ -1023,20 +1023,20 @@ pub struct TableShardStats {
    /// Highest WAL position the writer has seen. The difference against
    /// `replayAfterWalEntryPosition` is the WAL lag.
    pub wal_entry_position_last_seen: i64,
-    /// SSTables not yet merged into the base table.
-    pub sstables: Vec<SsTableStats>,
-    /// Whether a pass owns this table shard's compaction latch right now. Says *a*
+    /// Flushed L0 generations not yet merged into the base table.
+    pub generations: Vec<GenerationStats>,
+    /// Whether a pass owns this bucket's compaction latch right now. Says *a*
    /// driver is running, not *whose*, and the latch is held from dispatch —
    /// including while the pass queues for a pod-wide compactor permit. Read it
    /// as "do not pile on", never as "mine is progressing".
    pub compacting: bool,
-    /// Oldest first, active last. Absent for a `"Sealed"` table shard, whose
+    /// Oldest first, active last. Absent for a `"Sealed"` bucket, whose
    /// in-memory state is torn down.
    pub memtables: Option<Vec<MemtableStats>>,
 }

-impl From<lancedb::table::TableShardStats> for TableShardStats {
-    fn from(b: lancedb::table::TableShardStats) -> Self {
+impl From<lancedb::table::BucketStats> for BucketStats {
+    fn from(b: lancedb::table::BucketStats) -> Self {
        Self {
            shard_id: b.shard_id,
            status: b.status,
@@ -1045,7 +1045,7 @@ impl From<lancedb::table::TableShardStats> for TableShardStats {
            current_generation: b.current_generation as i64,
            replay_after_wal_entry_position: b.replay_after_wal_entry_position as i64,
            wal_entry_position_last_seen: b.wal_entry_position_last_seen as i64,
-            sstables: b.sstables.into_iter().map(Into::into).collect(),
+            generations: b.generations.into_iter().map(Into::into).collect(),
            compacting: b.compacting,
            memtables: b
                .memtables
@@ -1054,21 +1054,21 @@ impl From<lancedb::table::TableShardStats> for TableShardStats {
    }
 }

-/// Live per-table-shard LSM state, as returned by `Table#getLsmStats`.
+/// Live per-bucket LSM state, as returned by `Table#getLsmStats`.
 ///
-/// Nothing here is derived: sums and differences (total SSTable bytes, WAL lag) are
+/// Nothing here is derived: sums and differences (total L0 bytes, WAL lag) are
 /// the caller's to compute.
 #[napi(object)]
 #[derive(Clone, Debug)]
 pub struct LsmStats {
-    /// One entry per table shard backing this table.
-    pub table_shards: Vec<TableShardStats>,
+    /// One entry per bucket backing this table.
+    pub buckets: Vec<BucketStats>,
 }

 impl From<lancedb::table::LsmStats> for LsmStats {
    fn from(stats: lancedb::table::LsmStats) -> Self {
        Self {
-            table_shards: stats.table_shards.into_iter().map(Into::into).collect(),
+            buckets: stats.buckets.into_iter().map(Into::into).collect(),
        }
    }
 }
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.38.0-beta.11"
+version = "0.38.0-beta.12"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -6,7 +6,7 @@ import importlib.metadata
 import os
 from concurrent.futures import ThreadPoolExecutor
 from datetime import timedelta
-from typing import Dict, Optional, Union, Any, List, Iterable
+from typing import Dict, Optional, Union, Any, List, Iterable, TYPE_CHECKING

 __version__ = importlib.metadata.version("lancedb")

@@ -20,7 +20,7 @@ from .db import AsyncConnection, DBConnection, LanceDBConnection
 from .remote import ClientConfig
 from .remote.db import RemoteDBConnection
 from .expr import Expr, col, lit, func
-from .schema import blob, vector, BlobType
+from .schema import blob, vector
 from .job import AsyncJob, Job
 from .functions import (
    FunctionArtifactRequest as FunctionArtifactRequest,
@@ -38,7 +38,7 @@ from .materialized_view import (
    MaterializedView,
    MaterializedViewDefinition,
 )
-from .table import AsyncTable, Table
+from .table import AsyncTable, CompactionOptions, Table
 from .types import BaseTokenizerType
 from ._lancedb import Session
 from .namespace import (
@@ -49,6 +49,19 @@ from .namespace import (
 )


+if TYPE_CHECKING:
+    from lance.blob import BlobType as BlobType
+
+
+def __getattr__(name: str):
+    if name == "BlobType":
+        from .schema import BlobType
+
+        globals()["BlobType"] = BlobType
+        return BlobType
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
 def _check_s3_bucket_with_dots(
    uri: str, storage_options: Optional[Dict[str, str]]
 ) -> None:
@@ -545,6 +558,7 @@ __all__ = [
    "AsyncJob",
    "AsyncLanceNamespaceDBConnection",
    "AsyncTable",
+    "CompactionOptions",
    "FtsToken",
    "col",
    "Expr",
@@ -12,7 +12,7 @@ from typing import TYPE_CHECKING, Optional, Union
 import pyarrow as pa

 from .expr import Expr
-from .schema import blob_v2_column_paths
+from .schema import row_addressable_blob_v2_paths
 from .types import BlobMode, QueryProjection, QueryProjectionSpec

 if TYPE_CHECKING:
@@ -119,7 +119,7 @@ def blob_v2_projection_sources(
    schema: pa.Schema,
    projection: QueryProjection,
 ) -> dict[str, str]:
-    blob_columns = blob_v2_column_paths(schema)
+    blob_columns = row_addressable_blob_v2_paths(schema)
    if not blob_columns:
        return {}
    columns = set(blob_columns)
@@ -140,7 +140,9 @@ def v2_projection_needs_row_id(
 ) -> bool:
    if with_row_id:
        return False
-    return projection_includes_blob_column(projection, blob_v2_column_paths(schema))
+    return projection_includes_blob_column(
+        projection, row_addressable_blob_v2_paths(schema)
+    )


 def blob_auto_row_id_for_scan(
@@ -374,6 +374,7 @@ class Table:
        *,
        cleanup_since_ms: Optional[int] = None,
        delete_unverified: Optional[bool] = None,
+        compaction_options: Optional[Dict[str, Any]] = None,
    ) -> OptimizeStats: ...
    async def uri(self) -> str: ...
    async def initial_storage_options(self) -> Optional[Dict[str, str]]: ...
@@ -385,7 +386,7 @@ class Table:
    async def checkpoint_lsm(self) -> None: ...
    async def flush_lsm(self) -> None: ...
    async def compact_lsm(self) -> None: ...
-    async def get_lsm_stats(self, include_sstable_rows: bool) -> Optional[dict]: ...
+    async def get_lsm_stats(self, include_generation_rows: bool) -> Optional[dict]: ...
    async def close_lsm_writers(self) -> None: ...
    @property
    def tags(self) -> Tags: ...
@@ -67,7 +67,16 @@ from ..query import (
    LanceTakeQueryBuilder,
    LanceVectorQueryBuilder,
 )
-from ..table import AsyncTable, BlobMode, Branches, IndexStatistics, Query, Table, Tags
+from ..table import (
+    AsyncTable,
+    BlobMode,
+    Branches,
+    CompactionOptions,
+    IndexStatistics,
+    Query,
+    Table,
+    Tags,
+)
 from ..types import BaseTokenizerType


@@ -953,6 +962,7 @@ class RemoteTable(Table):
        *,
        cleanup_older_than: Optional[timedelta] = None,
        delete_unverified: bool = False,
+        compaction_options: Optional[CompactionOptions] = None,
    ):
        """
        optimize() is a no-op on LanceDB Cloud.
@@ -1029,11 +1039,11 @@ class RemoteTable(Table):
        [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
        return LOOP.run(self._table.compact_lsm())

-    def get_lsm_stats(self, *, include_sstable_rows: bool = False) -> Optional[dict]:
+    def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
        """Synchronous version of
        [`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
        return LOOP.run(
-            self._table.get_lsm_stats(include_sstable_rows=include_sstable_rows)
+            self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
        )

    def close_lsm_writers(self) -> None:
@@ -4,30 +4,34 @@

 """Schema helpers for Lance blob columns."""

+import importlib
+from typing import TYPE_CHECKING
+
 import pyarrow as pa
+import pyarrow.ipc
+
+if TYPE_CHECKING:
+    from lance.blob import BlobType as BlobType

 _BLOB_EXTENSION_NAME = "lance.blob.v2"
 _BLOB_V1_KEY = "lance-encoding:blob"
 _ARROW_EXT_NAME_KEY = "ARROW:extension:name"
+_BLOB_V2_STORAGE_TYPE = pa.struct(
+    [
+        pa.field("data", pa.large_binary(), nullable=True),
+        pa.field("uri", pa.utf8(), nullable=True),
+        pa.field("position", pa.uint64(), nullable=True),
+        pa.field("size", pa.uint64(), nullable=True),
+    ]
+)
+_resolved_blob_type = None


-class BlobType(pa.ExtensionType):
-    """PyArrow extension type for a Lance blob v2 column.
-
-    Queries return descriptors; call :meth:`~lancedb.table.Table.fetch_blob_files`
-    for lazy reads or :meth:`~lancedb.table.Table.fetch_blobs` for eager bytes.
-    """
+class _FallbackBlobType(pa.ExtensionType):
+    """lance.blob.v2 extension type used when pylance is not installed."""

    def __init__(self) -> None:
-        storage_type = pa.struct(
-            [
-                pa.field("data", pa.large_binary(), nullable=True),
-                pa.field("uri", pa.utf8(), nullable=True),
-                pa.field("position", pa.uint64(), nullable=True),
-                pa.field("size", pa.uint64(), nullable=True),
-            ]
-        )
-        super().__init__(storage_type, _BLOB_EXTENSION_NAME)
+        pa.ExtensionType.__init__(self, _BLOB_V2_STORAGE_TYPE, _BLOB_EXTENSION_NAME)

    def __arrow_ext_serialize__(self) -> bytes:
        return b""
@@ -35,23 +39,16 @@ class BlobType(pa.ExtensionType):
    @classmethod
    def __arrow_ext_deserialize__(
        cls, storage_type: pa.DataType, serialized: bytes
-    ) -> "BlobType":
+    ) -> "_FallbackBlobType":
        return cls()

    def __reduce__(self):
-        # Ensure pickle round-trips on older pyarrow (apache/arrow#35599).
        return type(self).__arrow_ext_deserialize__, (
            self.storage_type,
            self.__arrow_ext_serialize__(),
        )


-try:
-    pa.register_extension_type(BlobType())  # type: ignore[arg-type]
-except pa.ArrowKeyError:
-    pass
-
-
 def _metadata_value(metadata: dict, key: str):
    return metadata.get(key.encode()) or metadata.get(key)

@@ -92,43 +89,105 @@ def is_blob_like_field(field: pa.Field) -> bool:
    return is_blob_v2_field(field) or _metadata_marks_legacy_blob(field.metadata or {})


-def _collect_blob_paths(schema: pa.Schema, is_blob) -> list[str]:
-    paths: list[str] = []
+def _collect_blob_paths(schema: pa.Schema, is_blob) -> list[tuple[str, bool]]:
+    """Walk the schema and return (path, has_list_ancestor) for each blob field."""
+    paths: list[tuple[str, bool]] = []

-    def walk(fields, prefix: str) -> None:
+    def walk(fields, prefix: str, has_list_ancestor: bool) -> None:
        for field in fields:
            path = f"{prefix}.{field.name}" if prefix else field.name
            if is_blob(field):
-                paths.append(path)
+                paths.append((path, has_list_ancestor))
            elif pa.types.is_struct(field.type):
-                walk(field.type, path)
+                walk(field.type, path, has_list_ancestor)
            elif (
                pa.types.is_list(field.type)
                or pa.types.is_large_list(field.type)
                or pa.types.is_fixed_size_list(field.type)
            ):
-                walk([field.type.value_field], path)
+                walk([field.type.value_field], path, True)

-    walk(schema, "")
+    walk(schema, "", False)
    return paths


 def blob_column_paths(schema: pa.Schema) -> list[str]:
    """Dotted paths of blob-like columns (v2 extension or legacy metadata)."""
-    return _collect_blob_paths(schema, is_blob_like_field)
+    return [path for path, _ in _collect_blob_paths(schema, is_blob_like_field)]


 def blob_v2_column_paths(schema: pa.Schema) -> list[str]:
-    return _collect_blob_paths(schema, is_blob_v2_field)
+    return [path for path, _ in _collect_blob_paths(schema, is_blob_v2_field)]
+
+
+def row_addressable_blob_v2_paths(schema: pa.Schema) -> list[str]:
+    """Blob v2 paths with one blob addressable by table row id.
+
+    ``fetch_blobs`` and the descriptor row-id ride-along address one blob per
+    row, so a blob inside a list container has no row-id slot and no fetch
+    path. Those columns still store and query as raw descriptors.
+    """
+    return [
+        path
+        for path, has_list_ancestor in _collect_blob_paths(schema, is_blob_v2_field)
+        if not has_list_ancestor
+    ]


 def schema_has_blob_field(schema: pa.Schema) -> bool:
    return bool(blob_column_paths(schema))


+def _deserialize_registered_type(extension_type: pa.ExtensionType) -> pa.DataType:
+    """Return the type Arrow reconstructs for this extension name."""
+    schema = pa.schema([pa.field("value", extension_type)])
+    restored = pa.ipc.read_schema(schema.serialize())
+    return restored.field("value").type
+
+
+def _resolve_blob_type():
+    """Return the BlobType class this process should use.
+
+    pylance's class when it owns the lance.blob.v2 registry entry,
+    otherwise LanceDB's fallback. A different registered class is an error.
+    """
+    global _resolved_blob_type
+    if _resolved_blob_type is not None:
+        return _resolved_blob_type
+    try:
+        blob_module = importlib.import_module("lance.blob")
+    except ModuleNotFoundError as err:
+        if err.name not in ("lance", "lance.blob"):
+            raise
+    else:
+        blob_type = getattr(blob_module, "BlobType", None)
+        if blob_type is not None:
+            registered_type = _deserialize_registered_type(blob_type())
+            if type(registered_type) is not blob_type:
+                registered_cls = type(registered_type)
+                raise ValueError(
+                    "lance.blob.v2 is already registered by "
+                    f"{registered_cls.__module__}.{registered_cls.__qualname__}"
+                )
+            _resolved_blob_type = blob_type
+            return blob_type
+    try:
+        pa.register_extension_type(_FallbackBlobType())  # type: ignore[arg-type]
+    except pa.ArrowKeyError as err:
+        raise ValueError(
+            "lance.blob.v2 is already registered by another extension class"
+        ) from err
+    _resolved_blob_type = _FallbackBlobType
+    return _resolved_blob_type
+
+
 def blob(name: str, nullable: bool = True) -> pa.Field:
-    """Create a Lance blob v2 column field."""
-    return pa.field(name, BlobType(), nullable=nullable)
+    """Create a Lance blob v2 column field.
+
+    When pylance is installed this is ``lance.blob.BlobType``.
+    """
+    blob_type = _resolve_blob_type()
+    return pa.field(name, blob_type(), nullable=nullable)


 def vector(dimension: int, value_type: pa.DataType = pa.float32()) -> pa.DataType:
@@ -155,3 +214,11 @@ def vector(dimension: int, value_type: pa.DataType = pa.float32()) -> pa.DataTyp
    ... ])
    """
    return pa.list_(value_type, dimension)
+
+
+def __getattr__(name: str):
+    if name == "BlobType":
+        blob_type = _resolve_blob_type()
+        globals()["BlobType"] = blob_type
+        return blob_type
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -22,6 +22,7 @@ from typing import (
    Optional,
    Sequence,
    Tuple,
+    TypedDict,
    Union,
    overload,
 )
@@ -104,7 +105,12 @@ from .util import (
    value_to_sql,
 )
 from .index import lang_mapping
-from .schema import blob_v2_column_paths, schema_has_blob_field
+from .schema import (
+    blob_v2_column_paths,
+    is_blob_v2_field,
+    row_addressable_blob_v2_paths,
+    schema_has_blob_field,
+)


 def _should_push_down_query_table(
@@ -222,6 +228,88 @@ IndexConfigType = Union[
    FTS,
 ]

+
+class CompactionOptions(TypedDict, total=False):
+    """Options that control file compaction during table optimization.
+
+    Unspecified options use Lance's defaults.
+
+    Compaction planning is row based. Lowering ``target_rows_per_fragment``
+    based on the expected row size can bound later compaction passes once
+    oversized fragments have been rewritten. It does not split an existing
+    fragment, so the first pass over an oversized fragment is not subject to
+    that bound. ``max_bytes_per_file`` limits output file size, not compaction
+    memory. Source budgets keep whole planned tasks; if the first task exceeds
+    a budget, that run performs no compaction work.
+
+    Examples
+    --------
+    Derive a steady-state row target from the expected row size:
+
+    >>> desired_fragment_bytes = 750 * 1024 * 1024
+    >>> average_row_bytes = 1_500_000
+    >>> options: CompactionOptions = {
+    ...     "target_rows_per_fragment": max(
+    ...         1, desired_fragment_bytes // average_row_bytes
+    ...     ),
+    ... }
+    >>> await table.optimize(compaction_options=options)  # doctest: +SKIP
+    """
+
+    target_rows_per_fragment: int
+    """Target rows per fragment; existing oversized fragments are not split."""
+
+    max_rows_per_group: int
+    """Maximum number of rows per row group (default: 1,024)."""
+
+    max_bytes_per_file: Optional[int]
+    """Maximum output data-file size; this does not bound compaction memory."""
+
+    materialize_deletions: bool
+    """Whether to rewrite fragments containing deleted rows (default: True)."""
+
+    materialize_deletions_threshold: float
+    """Minimum deleted-row fraction that makes a fragment eligible (default: 0.1)."""
+
+    num_threads: Optional[int]
+    """Number of compaction tasks to run in parallel."""
+
+    batch_size: Optional[int]
+    """Number of rows per input scan batch."""
+
+    io_buffer_size: Optional[int]
+    """Maximum number of bytes queued in the input scan I/O buffer."""
+
+    defer_index_remap: bool
+    """Whether to defer index remapping during compaction (default: False)."""
+
+    index_remap_mode: Literal["direct", "compact"]
+    """How to construct the old-to-new row-address mapping."""
+
+    compaction_mode: Optional[
+        Literal["reencode", "try_binary_copy", "force_binary_copy"]
+    ]
+    """Whether compaction re-encodes data or uses binary copying."""
+
+    binary_copy_read_batch_bytes: Optional[int]
+    """Number of bytes read per batch during binary-copy compaction."""
+
+    max_source_fragments: Optional[int]
+    """Maximum number of source fragments compacted in one run."""
+
+    max_source_rows: Optional[int]
+    """Maximum live source rows per run, applied to whole planned tasks."""
+
+    max_source_bytes: Optional[int]
+    """Maximum source bytes per run, applied to whole planned tasks."""
+
+    excluded_fragment_ids: List[int]
+    """Fragment IDs to leave unchanged and use as planning boundaries."""
+
+    max_overlays_per_fragment: Optional[int]
+    """Maximum overlays before a fragment is fully compacted."""
+
+
 # Known distance metrics for legacy API detection
 KNOWN_METRICS = {"l2", "cosine", "dot", "hamming"}

@@ -426,6 +514,7 @@ def _cast_to_target_schema(

    def gen():
        for batch in reader:
+            batch = _coerce_blob_write_columns(batch, reordered_schema)
            # Table but not RecordBatch has cast.
            cast_batches = (
                pa.Table.from_batches([batch]).cast(reordered_schema).to_batches()
@@ -438,6 +527,166 @@ def _cast_to_target_schema(
    return pa.RecordBatchReader.from_batches(reordered_schema, gen())


+def _coerce_blob_write_columns(
+    batch: pa.RecordBatch, target_schema: pa.Schema
+) -> pa.RecordBatch:
+    """Materialize blob storage structs before the stream leaves Python.
+
+    merge_insert requires its source reader to already match the table's
+    physical schema. Unlike add and insert, it does not pass through
+    LanceDB's Rust blob coercion, so preserving binary input here would
+    reach Lance as binary and fail the schema check.
+    """
+    columns = []
+    fields = []
+    changed = False
+    for field, column in zip(batch.schema, batch.columns):
+        target_field = target_schema.field(field.name)
+        coerced = _coerce_blob_value(column, target_field)
+        if coerced is not column:
+            column = coerced
+            field = pa.field(
+                field.name,
+                coerced.type,
+                field.nullable,
+                target_field.metadata,
+            )
+            changed = True
+        columns.append(column)
+        fields.append(field)
+    if not changed:
+        return batch
+    return pa.RecordBatch.from_arrays(
+        columns, schema=pa.schema(fields, metadata=batch.schema.metadata)
+    )
+
+
+def _coerce_blob_value(column: pa.Array, target_field: pa.Field) -> pa.Array:
+    if is_blob_v2_field(target_field) and _can_coerce_to_blob(column.type):
+        return _coerce_value_to_blob(column, target_field)
+
+    target_type = target_field.type
+    if pa.types.is_struct(target_type) and pa.types.is_struct(column.type):
+        children = []
+        fields = []
+        changed = False
+        for source_field in column.type:
+            source_column = column.field(source_field.name)
+            nested_target = next(
+                (field for field in target_type if field.name == source_field.name),
+                None,
+            )
+            if nested_target is None:
+                children.append(source_column)
+                fields.append(source_field)
+                continue
+            coerced = _coerce_blob_value(source_column, nested_target)
+            if coerced is not source_column:
+                changed = True
+            child_array, child_type = _physical_array_and_type(coerced)
+            children.append(child_array)
+            fields.append(
+                pa.field(
+                    source_field.name,
+                    child_type,
+                    source_field.nullable,
+                    nested_target.metadata,
+                )
+            )
+        if not changed:
+            return column
+        return pa.StructArray.from_arrays(
+            children,
+            fields=fields,
+            mask=column.is_null() if column.null_count else None,
+        )
+
+    if _is_list_like(target_type) and _is_list_like(column.type):
+        return _coerce_blob_list_values(column, target_type.value_field)
+
+    return column
+
+
+def _coerce_blob_list_values(
+    column: pa.Array, target_value_field: pa.Field
+) -> pa.Array:
+    """Coerce blob values inside a list column, preserving offsets and nulls.
+
+    Works on the raw child values window instead of ``pc.list_flatten`` because
+    flatten drops values spanned by null slots, which would misalign offsets.
+    """
+    mask = column.is_null() if column.null_count else None
+    if pa.types.is_fixed_size_list(column.type):
+        list_size = column.type.list_size
+        values = column.values.slice(column.offset * list_size, len(column) * list_size)
+        coerced = _coerce_blob_value(values, target_value_field)
+        if coerced is values:
+            return column
+        physical_values, _ = _physical_array_and_type(coerced)
+        return pa.FixedSizeListArray.from_arrays(physical_values, list_size, mask=mask)
+    offsets = column.offsets
+    first_offset = offsets[0].as_py()
+    values = column.values.slice(
+        first_offset,
+        offsets[-1].as_py() - first_offset,
+    )
+    coerced = _coerce_blob_value(values, target_value_field)
+    if coerced is values:
+        return column
+    physical_values, _ = _physical_array_and_type(coerced)
+    if first_offset:
+        offsets = pc.subtract(offsets, pa.scalar(first_offset, offsets.type))
+    if pa.types.is_large_list(column.type):
+        return pa.LargeListArray.from_arrays(offsets, physical_values, mask=mask)
+    return pa.ListArray.from_arrays(offsets, physical_values, mask=mask)
+
+
+def _coerce_value_to_blob(values: pa.Array, target_field: pa.Field) -> pa.Array:
+    if pa.types.is_null(values.type):
+        data = pa.nulls(len(values), type=pa.large_binary())
+    elif pa.types.is_large_binary(values.type):
+        data = values
+    else:
+        data = values.cast(pa.large_binary())
+    length = len(values)
+    storage_type = target_field.type
+    if isinstance(storage_type, pa.ExtensionType):
+        storage_type = storage_type.storage_type
+    storage_fields = list(storage_type)
+    children = []
+    for storage_field in storage_fields:
+        if storage_field.name == "data":
+            children.append(data)
+        else:
+            children.append(pa.nulls(length, type=storage_field.type))
+    storage = pa.StructArray.from_arrays(
+        children,
+        fields=storage_fields,
+        mask=values.is_null() if values.null_count else None,
+    )
+    if isinstance(target_field.type, pa.ExtensionType):
+        return pa.ExtensionArray.from_storage(target_field.type, storage)
+    return storage
+
+
+def _physical_array_and_type(array: pa.Array) -> tuple[pa.Array, pa.DataType]:
+    if isinstance(array.type, pa.ExtensionType):
+        return array.storage, array.type.storage_type
+    return array, array.type
+
+
+def _can_coerce_to_blob(data_type: pa.DataType) -> bool:
+    return _is_binary_like(data_type) or pa.types.is_null(data_type)
+
+
+def _is_binary_like(data_type: pa.DataType) -> bool:
+    return (
+        pa.types.is_binary(data_type)
+        or pa.types.is_large_binary(data_type)
+        or pa.types.is_binary_view(data_type)
+    )
+
+
 def _field_extension_name(field: pa.Field) -> Optional[str]:
    extension_name = getattr(field.type, "extension_name", None)
    if extension_name is not None:
@@ -464,63 +713,71 @@ def _align_field_types(
        target_field = next((f for f in target_fields if f.name == field.name), None)
        if target_field is None:
            raise ValueError(f"Field '{field.name}' not found in target schema")
-        # Preserve arrow.json input until it reaches Lance. LanceDB exposes stored
-        # JSON columns as lance.json (JSONB-backed LargeBinary), but casting the
-        # input to that storage type here merely relabels the raw JSON bytes as
-        # JSONB. Lance must see arrow.json so it can perform the JSONB encoding.
-        if (
-            _field_extension_name(field) == "arrow.json"
-            and _field_extension_name(target_field) == "lance.json"
-        ):
-            new_fields.append(field)
-            continue
-        if pa.types.is_struct(target_field.type):
-            if pa.types.is_struct(field.type):
-                new_type = pa.struct(
-                    _align_field_types(
-                        field.type.fields,
-                        target_field.type.fields,
-                    )
+        new_fields.append(_align_field(field, target_field))
+    return new_fields
+
+
+def _align_list_value_field(
+    value_field: pa.Field, target_value_field: pa.Field
+) -> pa.Field:
+    # A list has exactly one child, so the inferred child name ("item") aligns
+    # positionally and adopts the table's child name; pa.Table.cast renames it.
+    return _align_field(value_field, target_value_field).with_name(
+        target_value_field.name
+    )
+
+
+def _align_field(field: pa.Field, target_field: pa.Field) -> pa.Field:
+    # Preserve arrow.json input until it reaches Lance. LanceDB exposes stored
+    # JSON columns as lance.json (JSONB-backed LargeBinary), but casting the
+    # input to that storage type here merely relabels the raw JSON bytes as
+    # JSONB. Lance must see arrow.json so it can perform the JSONB encoding.
+    if (
+        _field_extension_name(field) == "arrow.json"
+        and _field_extension_name(target_field) == "lance.json"
+    ):
+        return field
+    if pa.types.is_struct(target_field.type):
+        if pa.types.is_struct(field.type):
+            new_type = pa.struct(
+                _align_field_types(
+                    field.type.fields,
+                    target_field.type.fields,
                )
-            else:
-                new_type = target_field.type
-        elif pa.types.is_list(target_field.type):
-            if _is_list_like(field.type):
-                new_type = pa.list_(
-                    _align_field_types(
-                        [field.type.value_field],
-                        [target_field.type.value_field],
-                    )[0]
-                )
-            else:
-                new_type = target_field.type
-        elif pa.types.is_large_list(target_field.type):
-            if _is_list_like(field.type):
-                new_type = pa.large_list(
-                    _align_field_types(
-                        [field.type.value_field],
-                        [target_field.type.value_field],
-                    )[0]
-                )
-            else:
-                new_type = target_field.type
-        elif pa.types.is_fixed_size_list(target_field.type):
-            if _is_list_like(field.type):
-                new_type = pa.list_(
-                    _align_field_types(
-                        [field.type.value_field],
-                        [target_field.type.value_field],
-                    )[0],
-                    target_field.type.list_size,
-                )
-            else:
-                new_type = target_field.type
+            )
        else:
            new_type = target_field.type
-        new_fields.append(
-            pa.field(field.name, new_type, field.nullable, target_field.metadata)
-        )
-    return new_fields
+    elif pa.types.is_list(target_field.type):
+        if _is_list_like(field.type):
+            new_type = pa.list_(
+                _align_list_value_field(
+                    field.type.value_field, target_field.type.value_field
+                )
+            )
+        else:
+            new_type = target_field.type
+    elif pa.types.is_large_list(target_field.type):
+        if _is_list_like(field.type):
+            new_type = pa.large_list(
+                _align_list_value_field(
+                    field.type.value_field, target_field.type.value_field
+                )
+            )
+        else:
+            new_type = target_field.type
+    elif pa.types.is_fixed_size_list(target_field.type):
+        if _is_list_like(field.type):
+            new_type = pa.list_(
+                _align_list_value_field(
+                    field.type.value_field, target_field.type.value_field
+                ),
+                target_field.type.list_size,
+            )
+        else:
+            new_type = target_field.type
+    else:
+        new_type = target_field.type
+    return pa.field(field.name, new_type, field.nullable, target_field.metadata)


 def _infer_subschema(
@@ -589,7 +846,7 @@ def sanitize_create_table(
        schema = data.schema
    else:
        if schema is not None:
-            data = pa.Table.from_pylist([], schema)
+            data = pa.Table.from_batches([], schema=schema)
    if schema is None:
        if data is None:
            raise ValueError("Either data or schema must be provided")
@@ -1875,6 +2132,7 @@ class Table(ABC):
        cleanup_older_than: Optional[timedelta] = None,
        delete_unverified: bool = False,
        retrain: bool = False,
+        compaction_options: Optional[CompactionOptions] = None,
    ):
        """
        Optimize the on-disk data and indices for better performance.
@@ -1907,6 +2165,11 @@ class Table(ABC):

        retrain: bool, default False
            This parameter is no longer used and is deprecated.
+        compaction_options: CompactionOptions, optional
+            Options that control file compaction. For large rows, derive a lower
+            ``target_rows_per_fragment`` from the expected row size to bound later
+            passes. This does not cap the first pass over an existing oversized
+            fragment; see [CompactionOptions][lancedb.table.CompactionOptions].

        Notes
        -----
@@ -1991,9 +2254,11 @@ class Table(ABC):
            Function columns are supported only on LanceDB Cloud and
            Enterprise.
        computed: Dict[str, str], optional
-            A map of column name to a SQL expression defining the column. The
-            column's type and inputs are derived from the expression, so no
-            data type is supplied.
+            A mapping from output column names to SQL expressions derives each
+            output field from its expression. A direct projection of a Blob v2
+            field inherits Blob v2 semantics; other expressions derive their
+            ordinary Arrow type. Mapping order is declaration and dependency
+            order.

            Unlike ``transforms``, the expression is stored rather than
            evaluated now: the column is committed with no values, and rows get
@@ -2698,7 +2963,7 @@ class LanceTable(Table):
            arrow_tbl = self.to_arrow()
            if blob_mode == "descriptions":
                arrow_tbl = strip_auto_row_ids(
-                    arrow_tbl, blob_v2_column_paths(self.schema)
+                    arrow_tbl, row_addressable_blob_v2_paths(self.schema)
                )
            return arrow_tbl.to_pandas(**kwargs)

@@ -4025,6 +4290,7 @@ class LanceTable(Table):
        cleanup_older_than: Optional[timedelta] = None,
        delete_unverified: bool = False,
        retrain: bool = False,
+        compaction_options: Optional[CompactionOptions] = None,
    ):
        """
        Optimize the on-disk data and indices for better performance.
@@ -4057,6 +4323,11 @@ class LanceTable(Table):

        retrain: bool, default False
            This parameter is no longer used and is deprecated.
+        compaction_options: CompactionOptions, optional
+            Options that control file compaction. For large rows, derive a lower
+            ``target_rows_per_fragment`` from the expected row size to bound later
+            passes. This does not cap the first pass over an existing oversized
+            fragment; see [CompactionOptions][lancedb.table.CompactionOptions].

        Notes
        -----
@@ -4071,6 +4342,7 @@ class LanceTable(Table):
                cleanup_older_than=cleanup_older_than,
                delete_unverified=delete_unverified,
                retrain=retrain,
+                compaction_options=compaction_options,
            )
        )

@@ -4189,11 +4461,11 @@ class LanceTable(Table):
        [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
        return LOOP.run(self._table.compact_lsm())

-    def get_lsm_stats(self, *, include_sstable_rows: bool = False) -> Optional[dict]:
+    def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
        """Synchronous version of
        [`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
        return LOOP.run(
-            self._table.get_lsm_stats(include_sstable_rows=include_sstable_rows)
+            self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
        )

    def close_lsm_writers(self) -> None:
@@ -4916,16 +5188,16 @@ class AsyncTable:
    async def checkpoint_lsm(self) -> None:
        """Converge this table's LSM write path into its base table.

-        One flush, freezing every memtable into an SSTable, then compaction triggers
+        One flush, sealing every memtable into L0, then compaction triggers
        until every generation that existed at that moment has reached base.
        The loop runs client-side, reading progress from ``get_lsm_stats``.

-        Best-effort: SSTables created *while* it runs are deliberately not
+        Best-effort: generations created *while* it runs are deliberately not
        waited on, which is what lets it terminate on a table taking writes.
        Idempotent and safe on a cadence.

        There is no deadline, and the caller owns that. It returns when the
-        target SSTables are gone, raises on a terminal server fault, and
+        target generations are gone, raises on a terminal server fault, and
        otherwise waits however long the server takes. A slow table and a
        stuck one are the same picture from the client: the compactor pool is
        shared across every table on the node, so a checkpoint queued behind
@@ -4936,25 +5208,25 @@ class AsyncTable:
        await self._inner.checkpoint_lsm()

    async def flush_lsm(self) -> None:
-        """Freeze every table shard's active memtable into an SSTable.
+        """Seal every bucket's active memtable into L0.

-        Does not touch the base table — compacting SSTables into base is
+        Does not touch the base table — moving L0 into base is
        `compact_lsm`. On a node that has not claimed this table, this claims
        it and replays its WAL log first.
        """
        await self._inner.flush_lsm()

    async def compact_lsm(self) -> None:
-        """Trigger a background SSTable compaction pass per table shard.
+        """Trigger a background L0 to base compaction pass per bucket.

        Returns once the passes are dispatched, not once they finish: watch
        ``get_lsm_stats`` for progress, or use ``checkpoint_lsm`` to loop
-        until the current SSTables have reached base.
+        until the current L0 has reached base.
        """
        await self._inner.compact_lsm()

    async def get_lsm_stats(
-        self, *, include_sstable_rows: bool = False
+        self, *, include_generation_rows: bool = False
    ) -> Optional[dict]:
        """Read live per-bucket LSM state.

@@ -4967,12 +5239,12 @@ class AsyncTable:

        Parameters
        ----------
-        include_sstable_rows
-            Report a row count per SSTable. Off by default: each count
+        include_generation_rows
+            Report a row count per L0 generation. Off by default: each count
            opens an uncached Lance dataset, and ``checkpoint_lsm`` polls this
            needing only generation numbers.
        """
-        return await self._inner.get_lsm_stats(include_sstable_rows)
+        return await self._inner.get_lsm_stats(include_generation_rows)

    async def close_lsm_writers(self) -> None:
        """Drain and close any cached MemWAL shard writers for this table.
@@ -5102,7 +5374,9 @@ class AsyncTable:
        if blob_mode == "descriptions" or not schema_has_blob_field(schema):
            arrow_tbl = await self.to_arrow()
            if blob_mode == "descriptions":
-                arrow_tbl = strip_auto_row_ids(arrow_tbl, blob_v2_column_paths(schema))
+                arrow_tbl = strip_auto_row_ids(
+                    arrow_tbl, row_addressable_blob_v2_paths(schema)
+                )
            return arrow_tbl.to_pandas(**kwargs)

        if blob_mode == "lazy" and get_uri_scheme(await self.uri()) == "memory":
@@ -6092,8 +6366,11 @@ class AsyncTable:
            Function columns are supported only on LanceDB Cloud and
            Enterprise.
        computed: Dict[str, str], optional
-            A map of column name to a SQL expression defining the column. The
-            column's type and inputs are derived from the expression.
+            A mapping from output column names to SQL expressions derives each
+            output field from its expression. A direct projection of a Blob v2
+            field inherits Blob v2 semantics; other expressions derive their
+            ordinary Arrow type. Mapping order is declaration and dependency
+            order.

            Unlike ``transforms``, the expression is stored rather than
            evaluated now: the column is committed with no values, and rows get
@@ -6467,6 +6744,7 @@ class AsyncTable:
        cleanup_older_than: Optional[timedelta] = None,
        delete_unverified: bool = False,
        retrain=False,
+        compaction_options: Optional[CompactionOptions] = None,
    ) -> OptimizeStats:
        """
        Optimize the on-disk data and indices for better performance.
@@ -6499,6 +6777,11 @@ class AsyncTable:

        retrain: bool, default False
            This parameter is no longer used and is deprecated.
+        compaction_options: CompactionOptions, optional
+            Options that control file compaction. For large rows, derive a lower
+            ``target_rows_per_fragment`` from the expected row size to bound later
+            passes. This does not cap the first pass over an existing oversized
+            fragment; see [CompactionOptions][lancedb.table.CompactionOptions].

        Notes
        -----
@@ -6525,6 +6808,7 @@ class AsyncTable:
        return await self._inner.optimize(
            cleanup_since_ms=cleanup_since_ms,
            delete_unverified=delete_unverified,
+            compaction_options=compaction_options,
        )

    async def list_indices(self) -> Iterable[IndexConfig]:
@@ -2,10 +2,15 @@
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors

 import io
+import subprocess
+import sys
+import textwrap

+import lance
 import pyarrow as pa
 import pyarrow.compute as pc
 import pytest
+from lance.blob import BlobType as LanceBlobType

 import lancedb
 from lancedb._blob import (
@@ -18,6 +23,20 @@ from lancedb.index import FTS
 from lancedb.schema import blob_column_paths, blob_v2_column_paths


+_HIDE_LANCE_BLOB = """\
+import importlib.abc
+import sys
+
+class _MissingLanceBlob(importlib.abc.MetaPathFinder):
+    def find_spec(self, fullname, path, target=None):
+        if fullname == "lance.blob" or fullname.startswith("lance.blob."):
+            raise ModuleNotFoundError(fullname, name="lance.blob")
+
+sys.modules.pop("lance.blob", None)
+sys.meta_path.insert(0, _MissingLanceBlob())
+"""
+
+
 def _blob_table(name, rows):
    db = lancedb.connect("memory:///")
    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
@@ -51,6 +70,181 @@ def test_blob_factory_declares_v2_field():
    field = lancedb.blob("image")
    assert isinstance(field.type, pa.ExtensionType)
    assert field.type.extension_name == "lance.blob.v2"
+    assert lancedb.BlobType is LanceBlobType
+    assert type(field.type) is LanceBlobType
+
+
+def test_blob_type_works_without_pylance():
+    script = _HIDE_LANCE_BLOB + textwrap.dedent(
+        """\
+        import lancedb
+        import pyarrow as pa
+
+        field = lancedb.blob("image")
+        if not isinstance(field.type, pa.ExtensionType):
+            raise SystemExit("expected an extension type")
+        if field.type.extension_name != "lance.blob.v2":
+            raise SystemExit(field.type.extension_name)
+        if lancedb.BlobType is not type(field.type):
+            raise SystemExit("BlobType is not the field type class")
+        if lancedb.BlobType.__module__ != "lancedb.schema":
+            raise SystemExit(lancedb.BlobType.__module__)
+
+        db = lancedb.connect("memory:///")
+        table = db.create_table(
+            "images",
+            schema=pa.schema([pa.field("id", pa.int64()), field]),
+        )
+        table.add([{"id": 1, "image": b"hello"}])
+        result = (
+            table.merge_insert("id")
+            .when_matched_update_all()
+            .when_not_matched_insert_all()
+            .execute([{"id": 1, "image": b"updated"}, {"id": 2, "image": b"inserted"}])
+        )
+        if result.num_updated_rows != 1 or result.num_inserted_rows != 1:
+            raise SystemExit(
+                f"merge_insert rows updated={result.num_updated_rows} "
+                f"inserted={result.num_inserted_rows}"
+            )
+        """
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", script],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+
+
+def test_blob_resolves_pylance_type_without_eager_import():
+    script = textwrap.dedent(
+        """\
+        import sys
+        import lancedb
+
+        if "lance.blob" in sys.modules:
+            raise SystemExit("import lancedb imported lance.blob")
+        field = lancedb.blob("image")
+        from lance.blob import BlobType
+
+        if type(field.type) is not BlobType:
+            raise SystemExit(f"{type(field.type)} is not {BlobType}")
+        import lance
+
+        image = lance.blob_array([b"x"])
+        if type(image.type) is not BlobType:
+            raise SystemExit("blob_array used a different class")
+        if type(image.type) is not type(field.type):
+            raise SystemExit("field and array classes differ")
+        """
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", script],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+
+
+def test_blob_fallback_fails_if_name_already_registered():
+    script = _HIDE_LANCE_BLOB + textwrap.dedent(
+        """\
+        import pyarrow as pa
+
+        class OtherBlobType(pa.ExtensionType):
+            def __init__(self):
+                super().__init__(
+                    pa.struct([pa.field("data", pa.large_binary())]),
+                    "lance.blob.v2",
+                )
+
+            def __arrow_ext_serialize__(self):
+                return b""
+
+            @classmethod
+            def __arrow_ext_deserialize__(cls, storage_type, serialized):
+                return cls()
+
+        pa.register_extension_type(OtherBlobType())
+        import lancedb
+
+        try:
+            lancedb.blob("image")
+        except ValueError as err:
+            if "already registered" not in str(err):
+                raise SystemExit(err)
+        else:
+            raise SystemExit("expected ValueError")
+        """
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", script],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+
+
+def test_blob_type_rejects_competing_registration_with_pylance():
+    script = textwrap.dedent(
+        """\
+        import pyarrow as pa
+        import pyarrow.ipc
+
+        class OtherBlobType(pa.ExtensionType):
+            def __init__(self):
+                super().__init__(
+                    pa.struct(
+                        [
+                            pa.field("data", pa.large_binary()),
+                            pa.field("uri", pa.utf8()),
+                            pa.field("position", pa.uint64()),
+                            pa.field("size", pa.uint64()),
+                        ]
+                    ),
+                    "lance.blob.v2",
+                )
+
+            def __arrow_ext_serialize__(self):
+                return b""
+
+            @classmethod
+            def __arrow_ext_deserialize__(cls, storage_type, serialized):
+                return cls()
+
+        pa.register_extension_type(OtherBlobType())
+
+        from lance.blob import BlobType
+
+        if BlobType is OtherBlobType:
+            raise SystemExit("pylance BlobType was replaced")
+        schema = pa.schema([pa.field("value", BlobType())])
+        restored = pa.ipc.read_schema(schema.serialize())
+        if type(restored.field("value").type) is not OtherBlobType:
+            raise SystemExit(type(restored.field("value").type))
+
+        import lancedb
+
+        try:
+            lancedb.blob("image")
+        except ValueError as err:
+            if "__main__.OtherBlobType" not in str(err):
+                raise SystemExit(err)
+        else:
+            raise SystemExit("expected ValueError")
+        """
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", script],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr


 def test_blob_v2_column_paths_include_list_children():
@@ -203,6 +397,292 @@ def test_fetch_blobs_round_trip():
    assert [blobs[0].as_py(), blobs[1].as_py()] == [b"alpha", b"beta"]


+def test_merge_insert_writes_python_bytes():
+    table = _blob_table("merge_bytes", [{"id": 1, "image": b"before"}])
+    result = (
+        table.merge_insert("id")
+        .when_matched_update_all()
+        .when_not_matched_insert_all()
+        .execute([{"id": 1, "image": b"updated"}, {"id": 2, "image": b"inserted"}])
+    )
+    assert result.num_updated_rows == 1
+    assert result.num_inserted_rows == 1
+    by_id = _row_ids_by_id(table)
+    blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
+    assert blobs.to_pylist() == [b"updated", b"inserted"]
+
+
+def test_merge_insert_bytes_after_reopen_without_touching_blob_type(tmp_path):
+    db = lancedb.connect(tmp_path)
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    table = db.create_table("images", schema=schema)
+    table.add([{"id": 1, "image": b"hello"}])
+
+    script = textwrap.dedent(
+        f"""\
+        import lancedb
+
+        db = lancedb.connect({str(tmp_path)!r})
+        table = db.open_table("images")
+        image_type = table.schema.field("image").type
+        if type(image_type).__name__ != "StructType":
+            raise SystemExit(f"expected StructType, got {{type(image_type)}}")
+        result = (
+            table.merge_insert("id")
+            .when_matched_update_all()
+            .when_not_matched_insert_all()
+            .execute(
+                [{{"id": 1, "image": b"updated"}}, {{"id": 2, "image": b"inserted"}}]
+            )
+        )
+        if result.num_updated_rows != 1 or result.num_inserted_rows != 1:
+            raise SystemExit(
+                f"rows updated={{result.num_updated_rows}} "
+                f"inserted={{result.num_inserted_rows}}"
+            )
+        hits = table.search().with_row_id(True).limit(10).to_arrow()
+        by_id = dict(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))
+        blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
+        if blobs.to_pylist() != [b"updated", b"inserted"]:
+            raise SystemExit(blobs.to_pylist())
+        """
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", script],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+
+
+def test_merge_insert_bytes_after_reopen_without_pylance(tmp_path):
+    db = lancedb.connect(tmp_path)
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    table = db.create_table("images", schema=schema)
+    table.add([{"id": 1, "image": b"hello"}])
+
+    script = _HIDE_LANCE_BLOB + textwrap.dedent(
+        f"""\
+        import lancedb
+
+        db = lancedb.connect({str(tmp_path)!r})
+        table = db.open_table("images")
+        image_type = table.schema.field("image").type
+        if type(image_type).__name__ != "StructType":
+            raise SystemExit(f"expected StructType, got {{type(image_type)}}")
+        result = (
+            table.merge_insert("id")
+            .when_matched_update_all()
+            .when_not_matched_insert_all()
+            .execute(
+                [{{"id": 1, "image": b"updated"}}, {{"id": 2, "image": b"inserted"}}]
+            )
+        )
+        if result.num_updated_rows != 1 or result.num_inserted_rows != 1:
+            raise SystemExit(
+                f"rows updated={{result.num_updated_rows}} "
+                f"inserted={{result.num_inserted_rows}}"
+            )
+        hits = table.search().with_row_id(True).limit(10).to_arrow()
+        by_id = dict(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))
+        blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
+        if blobs.to_pylist() != [b"updated", b"inserted"]:
+            raise SystemExit(blobs.to_pylist())
+        """
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", script],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+
+
+def test_merge_insert_blob_array_into_reopened_unregistered_table(tmp_path):
+    db = lancedb.connect(tmp_path)
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    table = db.create_table("images", schema=schema)
+    table.add([{"id": 1, "image": b"before"}])
+
+    script = textwrap.dedent(
+        f"""\
+        import pyarrow as pa
+        import lancedb
+
+        db = lancedb.connect({str(tmp_path)!r})
+        table = db.open_table("images")
+        image_type = table.schema.field("image").type
+        if type(image_type).__name__ != "StructType":
+            raise SystemExit(
+                f"expected StructType before lance import, got {{type(image_type)}}"
+            )
+
+        import lance
+
+        updates = pa.Table.from_arrays(
+            [
+                pa.array([1, 2], type=pa.int64()),
+                lance.blob_array([b"updated", b"inserted"]),
+            ],
+            names=["id", "image"],
+        )
+        result = (
+            table.merge_insert("id")
+            .when_matched_update_all()
+            .when_not_matched_insert_all()
+            .execute(updates)
+        )
+        if result.num_updated_rows != 1 or result.num_inserted_rows != 1:
+            raise SystemExit(
+                f"rows updated={{result.num_updated_rows}} "
+                f"inserted={{result.num_inserted_rows}}"
+            )
+        hits = table.search().with_row_id(True).limit(10).to_arrow()
+        by_id = dict(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))
+        blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
+        if blobs.to_pylist() != [b"updated", b"inserted"]:
+            raise SystemExit(blobs.to_pylist())
+        """
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", script],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+
+
+def test_add_all_null_blob_column():
+    db = lancedb.connect("memory:///")
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    table = db.create_table("all_null", schema=schema)
+    table.add([{"id": 1, "image": None}, {"id": 2, "image": None}])
+    by_id = _row_ids_by_id(table)
+    blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
+    assert blobs.to_pylist() == [None, None]
+
+
+def test_create_table_nested_blob_schema_without_rows():
+    db = lancedb.connect("memory:///")
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("info", pa.struct([lancedb.blob("blob")])),
+            pa.field("images", pa.list_(lancedb.blob("image"))),
+        ]
+    )
+    table = db.create_table("nested_empty", schema=schema)
+    assert table.count_rows() == 0
+
+
+def test_merge_insert_nested_blob_dicts():
+    db = lancedb.connect("memory:///")
+    info = pa.StructArray.from_arrays(
+        [
+            pa.array(["first"], type=pa.string()),
+            _blob_array("blob", [b"before"]),
+        ],
+        names=["name", "blob"],
+    )
+    data = pa.Table.from_arrays(
+        [pa.array([1], type=pa.int64()), info],
+        names=["id", "info"],
+    )
+    table = db.create_table("nested_merge", data=data)
+    result = (
+        table.merge_insert("id")
+        .when_matched_update_all()
+        .execute([{"id": 1, "info": {"name": "first", "blob": b"after"}}])
+    )
+    assert result.num_updated_rows == 1
+    by_id = _row_ids_by_id(table)
+    blobs = table.fetch_blobs("info.blob", [by_id[1]])
+    assert blobs.to_pylist() == [b"after"]
+
+
+def _list_blob_table(name):
+    db = lancedb.connect("memory:///")
+    blob_field = lancedb.blob("image")
+    images = pa.ListArray.from_arrays(
+        pa.array([0, 1], type=pa.int32()), _blob_array("image", [b"before"])
+    )
+    data = pa.Table.from_arrays(
+        [pa.array([1], type=pa.int64()), images],
+        schema=pa.schema(
+            [pa.field("id", pa.int64()), pa.field("images", pa.list_(blob_field))]
+        ),
+    )
+    return db.create_table(name, data=data)
+
+
+def test_merge_insert_list_blob_dicts():
+    table = _list_blob_table("list_merge")
+    result = (
+        table.merge_insert("id")
+        .when_matched_update_all()
+        .when_not_matched_insert_all()
+        .execute([{"id": 1, "images": [b"one", b"two"]}, {"id": 2, "images": None}])
+    )
+    assert result.num_updated_rows == 1
+    assert result.num_inserted_rows == 1
+    hits = table.search().limit(10).to_arrow()
+    sizes = {
+        row["id"]: None if row["images"] is None else [d["size"] for d in row["images"]]
+        for row in hits.to_pylist()
+    }
+    assert sizes == {1: [3, 3], 2: None}
+
+
+def test_list_blob_column_queries_as_raw_descriptors():
+    table = _list_blob_table("list_query")
+    hits = table.search().limit(10).to_arrow()
+    element = hits.schema.field("images").type.value_type
+    assert pa.types.is_struct(element)
+    assert "_lance_row_id" not in element.names
+    with pytest.raises(ValueError, match="expected struct before segment"):
+        table.fetch_blobs("images.image", [0])
+
+
+def test_row_addressable_paths_exclude_list_children():
+    from lancedb.schema import row_addressable_blob_v2_paths
+
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("info", pa.struct([lancedb.blob("blob")])),
+            pa.field("images", pa.list_(lancedb.blob("image"))),
+        ]
+    )
+    assert blob_v2_column_paths(schema) == ["info.blob", "images.image"]
+    assert row_addressable_blob_v2_paths(schema) == ["info.blob"]
+
+
+def test_merge_insert_writes_pylance_blob_array():
+    table = _blob_table("merge_pylance", [{"id": 1, "image": b"before"}])
+    image = lance.blob_array([b"updated", b"inserted"])
+    assert type(image.type) is LanceBlobType
+    assert type(image.type) is type(lancedb.BlobType())
+    updates = pa.Table.from_arrays(
+        [pa.array([1, 2], type=pa.int64()), image], names=["id", "image"]
+    )
+
+    result = (
+        table.merge_insert("id")
+        .when_matched_update_all()
+        .when_not_matched_insert_all()
+        .execute(updates)
+    )
+
+    assert result.num_updated_rows == 1
+    assert result.num_inserted_rows == 1
+    by_id = _row_ids_by_id(table)
+    blobs = table.fetch_blobs("image", [by_id[1], by_id[2]])
+    assert blobs.to_pylist() == [b"updated", b"inserted"]
+
+
 def test_fetch_blobs_accepts_query_result():
    table = _blob_table("from_result", [{"id": 1, "image": b"gamma"}])
    hits = table.search().limit(10).to_arrow()
@@ -1278,9 +1278,9 @@ def test_get_lsm_stats_sync():
    with lsm_test_table(lsm_handler) as table:
        assert table.get_lsm_stats() == {"buckets": [bucket]}
        # Off by default, and forwarded when asked for.
-        assert seen_bodies == [{"include_sstable_rows": False}]
-        table.get_lsm_stats(include_sstable_rows=True)
-        assert seen_bodies[-1] == {"include_sstable_rows": True}
+        assert seen_bodies == [{"include_generation_rows": False}]
+        table.get_lsm_stats(include_generation_rows=True)
+        assert seen_bodies[-1] == {"include_generation_rows": True}


 def test_get_lsm_stats_sync_returns_none_when_lsm_disabled():
@@ -1309,7 +1309,7 @@ def test_flush_and_compact_lsm_sync():


 def test_checkpoint_lsm_sync():
-    """Freeze, read the watermark, and return once no SSTables remain.
+    """Seal, read the watermark, and return once L0 holds nothing.

    The convergence loop itself is covered in Rust; this pins the sync
    binding to the endpoints it drives.
@@ -1319,7 +1319,7 @@ def test_checkpoint_lsm_sync():
    def lsm_handler(request, route):
        called.append(route)
        if route == "get_lsm_stats":
-            # An empty SSTable tier yields no target watermark, so the loop is done
+            # An empty L0 yields no target watermark, so the loop is done
            # after the seal without ever polling compaction.
            send_json(request, {"lsm_stats": {"buckets": []}})
        else:
@@ -13,7 +13,7 @@ from concurrent.futures import ThreadPoolExecutor
 from datetime import date, datetime, timedelta
 from decimal import Decimal
 from time import sleep
-from typing import List
+from typing import Any, List
 from unittest.mock import patch

 import lancedb
@@ -3876,6 +3876,100 @@ async def test_optimize(mem_db_async: AsyncConnection):
    assert await table.query().to_arrow() == pa.table({"x": [[1], [2]]})


+@pytest.mark.asyncio
+async def test_optimize_compaction_options(mem_db_async: AsyncConnection):
+    table = await mem_db_async.create_table("test", data=[{"x": 1}])
+    await table.add([{"x": 2}])
+
+    stats = await table.optimize(
+        compaction_options={
+            "target_rows_per_fragment": 1,
+            "batch_size": 1,
+            "num_threads": 1,
+        }
+    )
+    assert stats.compaction.fragments_removed == 0
+    assert stats.compaction.fragments_added == 0
+
+    stats = await table.optimize(compaction_options={"target_rows_per_fragment": 3})
+    assert stats.compaction.fragments_removed == 2
+    assert stats.compaction.fragments_added == 1
+
+    with pytest.raises(ValueError, match="Invalid compaction option: unknown"):
+        await table.optimize(compaction_options={"unknown": 1})
+
+
+@pytest.mark.parametrize("option", ["max_source_rows", "max_source_bytes"])
+@pytest.mark.asyncio
+async def test_optimize_compaction_source_limits(
+    mem_db_async: AsyncConnection, option: str
+):
+    table = await mem_db_async.create_table("test", data=[{"x": 1}])
+    await table.add([{"x": 2}])
+
+    stats = await table.optimize(
+        compaction_options={
+            "target_rows_per_fragment": 3,
+            option: 1,
+        }
+    )
+    assert stats.compaction.fragments_removed == 0
+    assert stats.compaction.fragments_added == 0
+
+
+@pytest.mark.asyncio
+async def test_optimize_compaction_excluded_fragments(mem_db_async: AsyncConnection):
+    table = await mem_db_async.create_table("test", data=[{"x": 1}])
+    await table.add([{"x": 2}])
+
+    stats = await table.optimize(
+        compaction_options={
+            "target_rows_per_fragment": 3,
+            "excluded_fragment_ids": [0],
+        }
+    )
+    assert stats.compaction.fragments_removed == 0
+    assert stats.compaction.fragments_added == 0
+
+    stats = await table.optimize(compaction_options={"target_rows_per_fragment": 3})
+    assert stats.compaction.fragments_removed == 2
+    assert stats.compaction.fragments_added == 1
+
+
+@pytest.mark.parametrize(
+    ("option", "value", "message"),
+    [
+        ("target_rows_per_fragment", 0, "must be between 1 and 4294967295"),
+        ("max_rows_per_group", 0, "must be between 1 and 4294967295"),
+        ("batch_size", 0, "must be between 1 and 4294967295"),
+        ("num_threads", 0, "must be greater than 0"),
+        ("target_rows_per_fragment", 2**32, "must be between 1 and 4294967295"),
+        ("max_rows_per_group", 2**32, "must be between 1 and 4294967295"),
+        ("batch_size", 2**32, "must be between 1 and 4294967295"),
+        ("io_buffer_size", 2**63, "must be at most 9223372036854775807"),
+        ("max_source_rows", 0, "must be greater than 0"),
+        ("max_source_bytes", 0, "must be greater than 0"),
+        (
+            "excluded_fragment_ids",
+            [-1],
+            "must contain values between 0 and 4294967295",
+        ),
+        (
+            "excluded_fragment_ids",
+            [2**32],
+            "must contain values between 0 and 4294967295",
+        ),
+    ],
+)
+@pytest.mark.asyncio
+async def test_optimize_compaction_options_validation(
+    mem_db_async: AsyncConnection, option: str, value: Any, message: str
+):
+    table = await mem_db_async.create_table("test", data=[{"x": 1}])
+    with pytest.raises(ValueError, match=message):
+        await table.optimize(compaction_options={option: value})
+
+
@pytest.mark.asyncio
 async def test_optimize_delete_unverified(tmp_db_async: AsyncConnection, tmp_path):
    table = await tmp_db_async.create_table(
@@ -4087,6 +4181,29 @@ def test_computed_column_rejects_transforms_and_computed_together(tmp_path):
        table.add_columns({"a": "x + 1"}, computed={"b": "x * 2"})


+def test_computed_column_blob_projection_inherits_semantics(tmp_path):
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    db = lancedb.connect(tmp_path)
+    table = db.create_table("computed_column_blob", schema=schema)
+    table.add(
+        [
+            {"id": 1, "image": b"hello"},
+            {"id": 2, "image": b""},
+            {"id": 3, "image": None},
+        ]
+    )
+
+    table.add_columns(computed={"image_copy": "image", "second_copy": "image_copy"})
+    assert table.refresh_column("image_copy").rows_filled == 2
+    assert table.refresh_column("second_copy").rows_filled == 2
+    assert table.blob_columns() == ["image", "image_copy", "second_copy"]
+
+    hits = table.search().with_row_id(True).limit(10).to_arrow()
+    rows = sorted(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))
+    copied = table.fetch_blobs("second_copy", [row_id for _, row_id in rows])
+    assert copied.to_pylist() == [b"hello", b"", None]
+
+
@pytest.mark.asyncio
 async def test_computed_column_async(tmp_path):
    db = await lancedb.connect_async(tmp_path)
@@ -7,6 +7,7 @@ import pathlib
 from typing import Optional

 import lance
+from lance.blob import BlobType as LanceBlobType
 from lancedb.conftest import MockTextEmbeddingFunction
 from lancedb.embeddings.base import EmbeddingFunctionConfig
 from lancedb.embeddings.registry import EmbeddingFunctionRegistry
@@ -907,6 +908,165 @@ def test_cast_to_target_schema():
    assert output == expected


+def test_cast_to_target_schema_coerces_binary_to_blob_v2():
+    data = pa.table({"image": pa.array([b"hello", None], type=pa.binary())})
+    target = pa.schema([lancedb.blob("image")])
+
+    output = _cast_to_target_schema(data.to_reader(), target).read_all()
+
+    image = output["image"].chunk(0)
+    assert type(image.type) is lancedb.BlobType
+    assert image.storage.to_pylist() == [
+        {"data": b"hello", "uri": None, "position": None, "size": None},
+        None,
+    ]
+
+
+def test_cast_to_target_schema_coerces_binary_to_metadata_blob_struct():
+    storage = lancedb.blob("image").type.storage_type
+    target = pa.schema(
+        [
+            pa.field(
+                "image",
+                storage,
+                metadata={
+                    b"ARROW:extension:name": b"lance.blob.v2",
+                    b"ARROW:extension:metadata": b"",
+                },
+            )
+        ]
+    )
+    data = pa.table({"image": pa.array([b"hello", None], type=pa.binary())})
+
+    output = _cast_to_target_schema(data.to_reader(), target).read_all()
+
+    image = output["image"].chunk(0)
+    assert not isinstance(image.type, pa.ExtensionType)
+    assert image.to_pylist() == [
+        {"data": b"hello", "uri": None, "position": None, "size": None},
+        None,
+    ]
+
+
+def test_cast_to_target_schema_coerces_nested_binary_blob():
+    data = pa.table(
+        {
+            "info": pa.array(
+                [{"blob": b"hello"}, {"blob": None}],
+                type=pa.struct([pa.field("blob", pa.binary())]),
+            )
+        }
+    )
+    target = pa.schema([pa.field("info", pa.struct([lancedb.blob("blob")]))])
+
+    output = _cast_to_target_schema(data.to_reader(), target).read_all()
+
+    blob = output["info"].chunk(0).field("blob")
+    assert type(blob.type) is lancedb.BlobType
+    assert blob.storage.to_pylist() == [
+        {"data": b"hello", "uri": None, "position": None, "size": None},
+        None,
+    ]
+
+
+def test_cast_to_target_schema_coerces_list_binary_blob_with_inferred_child_name():
+    data = pa.table(
+        {"images": pa.array([[b"a", b"b"], None], type=pa.list_(pa.binary()))}
+    )
+    target = pa.schema([pa.field("images", pa.list_(lancedb.blob("image")))])
+
+    output = _cast_to_target_schema(data.to_reader(), target).read_all()
+
+    images = output["images"].chunk(0)
+    assert images.type.value_field.name == "image"
+    assert type(images.type.value_type) is lancedb.BlobType
+    assert images.to_pylist()[1] is None
+    assert images.values.storage.to_pylist() == [
+        {"data": b"a", "uri": None, "position": None, "size": None},
+        {"data": b"b", "uri": None, "position": None, "size": None},
+    ]
+
+
+def test_list_blob_coercion_preserves_null_slots_with_nonzero_extent():
+    child = pa.field("image", pa.binary())
+    source = pa.ListArray.from_arrays(
+        pa.array([0, 2, 4], type=pa.int32()),
+        pa.array([b"a", b"b", b"dead", b"beef"], type=pa.binary()),
+        mask=pa.array([False, True]),
+    ).cast(pa.list_(child))
+    target = pa.schema([pa.field("images", pa.list_(lancedb.blob("image")))])
+
+    output = _cast_to_target_schema(
+        pa.table({"images": source}).to_reader(), target
+    ).read_all()
+
+    images = output["images"].chunk(0)
+    assert images.to_pylist()[1] is None
+    assert [b["data"] for b in images.to_pylist()[0]] == [b"a", b"b"]
+
+
+def test_fixed_size_list_blob_coercion_keeps_null_rows():
+    child = pa.field("frame", pa.binary())
+    source = (
+        pa.FixedSizeListArray.from_arrays(
+            pa.array([b"a", b"b", b"c", b"d"], type=pa.binary()), 2
+        )
+        .take(pa.array([0, None], type=pa.int32()))
+        .cast(pa.list_(child, 2))
+    )
+    target = pa.schema([pa.field("frames", pa.list_(lancedb.blob("frame"), 2))])
+
+    output = _cast_to_target_schema(
+        pa.table({"frames": source}).to_reader(), target
+    ).read_all()
+
+    frames = output["frames"].chunk(0)
+    assert frames.to_pylist()[1] is None
+    assert [b["data"] for b in frames.to_pylist()[0]] == [b"a", b"b"]
+
+
+def test_cast_to_target_schema_accepts_pylance_blob_v2():
+    target_type = lancedb.BlobType()
+    source = lance.blob_array([b"hello", None])
+    assert type(source.type) is LanceBlobType
+    assert type(source.type) is type(target_type)
+    data = pa.table({"image": source})
+    target = pa.schema([pa.field("image", target_type)])
+
+    output = _cast_to_target_schema(data.to_reader(), target).read_all()
+
+    image = output["image"].chunk(0)
+    assert type(image.type) is LanceBlobType
+    assert image.type == target_type
+    assert image.storage.to_pylist() == [
+        {"data": b"hello", "uri": None, "position": None, "size": None},
+        None,
+    ]
+
+
+def test_cast_to_target_schema_rejects_different_blob_v2_class():
+    class OtherBlobType(pa.ExtensionType):
+        def __init__(self):
+            super().__init__(lancedb.BlobType().storage_type, "lance.blob.v2")
+
+        def __arrow_ext_serialize__(self) -> bytes:
+            return b""
+
+        @classmethod
+        def __arrow_ext_deserialize__(
+            cls, storage_type: pa.DataType, serialized: bytes
+        ) -> "OtherBlobType":
+            return cls()
+
+    storage = lance.blob_array([b"hello"]).storage
+    source = pa.ExtensionArray.from_storage(OtherBlobType(), storage)
+    data = pa.table({"image": source})
+    target = pa.schema([lancedb.blob("image")])
+
+    with pytest.raises(pa.ArrowTypeError, match="different extension type"):
+        _cast_to_target_schema(data.to_reader(), target).read_all()
+
+
 def test_sanitize_data_stream():
    # Make sure we don't collect the whole stream when running sanitize_data
    schema = pa.schema({"a": pa.int32()})
@@ -20,8 +20,9 @@ use arrow::{
 use lancedb::blob::{BlobFile, BlobRangeRequest};
 use lancedb::index::scalar::FtsIndexBuilder;
 use lancedb::table::{
-    AddDataMode, ColumnAlteration, Duration, FieldMetadataUpdate, FtsToken as LanceDbFtsToken,
-    NewColumnTransform, OptimizeAction, OptimizeOptions, Ref, Table as LanceDbTable,
+    AddDataMode, ColumnAlteration, CompactionMode, CompactionOptions, Duration,
+    FieldMetadataUpdate, FtsToken as LanceDbFtsToken, IndexRemapMode, NewColumnTransform,
+    OptimizeAction, OptimizeOptions, Ref, Table as LanceDbTable,
 };
 use lancedb::tokenize as lancedb_tokenize;
 use pyo3::{
@@ -33,16 +34,16 @@ use pyo3::{

 mod scannable;

-/// Convert `LsmStats` to a Python dict, preserving the per-table-shard list.
+/// Convert `LsmStats` to a Python dict, preserving the per-bucket list.
 ///
 /// Deliberately not flattened to a table-level summary: a table is N
-/// table shards on one node, and the per-shard detail is the reason the
-/// endpoint exists — flattening hides the single hot table shard someone opened
+/// buckets on one node, and the per-bucket detail is the reason the
+/// endpoint exists — flattening hides the single hot bucket someone opened
 /// it to find.
 fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult<Py<PyDict>> {
    let out = PyDict::new(py);
-    let table_shards = PyList::empty(py);
-    for b in &stats.table_shards {
+    let buckets = PyList::empty(py);
+    for b in &stats.buckets {
        let e = PyDict::new(py);
        e.set_item("shard_id", &b.shard_id)?;
        e.set_item("status", &b.status)?;
@@ -58,15 +59,15 @@ fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult
            b.wal_entry_position_last_seen,
        )?;

-        let sstables = PyList::empty(py);
-        for g in &b.sstables {
+        let generations = PyList::empty(py);
+        for g in &b.generations {
            let ge = PyDict::new(py);
            ge.set_item("generation", g.generation)?;
            ge.set_item("bytes", g.bytes)?;
            ge.set_item("rows", g.rows)?;
-            sstables.append(ge)?;
+            generations.append(ge)?;
        }
-        e.set_item("sstables", sstables)?;
+        e.set_item("generations", generations)?;
        e.set_item("compacting", b.compacting)?;

        e.set_item(
@@ -88,9 +89,9 @@ fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult
                })
                .transpose()?,
        )?;
-        table_shards.append(e)?;
+        buckets.append(e)?;
    }
-    out.set_item("table_shards", table_shards)?;
+    out.set_item("buckets", buckets)?;
    Ok(out.unbind())
 }

@@ -100,6 +101,128 @@ enum PredicateArg {
    Sql(String),
 }

+fn validate_positive_u32(value: u64, name: &str) -> PyResult<usize> {
+    if !(1..=u32::MAX as u64).contains(&value) {
+        return Err(PyValueError::new_err(format!(
+            "{name} must be between 1 and {}",
+            u32::MAX
+        )));
+    }
+    Ok(value as usize)
+}
+
+fn positive_u32(value: &Bound<'_, PyAny>, name: &str) -> PyResult<usize> {
+    validate_positive_u32(value.extract()?, name)
+}
+
+fn optional_positive_u32(value: &Bound<'_, PyAny>, name: &str) -> PyResult<Option<usize>> {
+    value
+        .extract::<Option<u64>>()?
+        .map(|value| validate_positive_u32(value, name))
+        .transpose()
+}
+
+fn optional_positive_usize(value: &Bound<'_, PyAny>, name: &str) -> PyResult<Option<usize>> {
+    let value: Option<usize> = value.extract()?;
+    if value == Some(0) {
+        return Err(PyValueError::new_err(format!(
+            "{name} must be greater than 0"
+        )));
+    }
+    Ok(value)
+}
+
+fn optional_positive_u64(value: &Bound<'_, PyAny>, name: &str) -> PyResult<Option<u64>> {
+    let value: Option<u64> = value.extract()?;
+    if value == Some(0) {
+        return Err(PyValueError::new_err(format!(
+            "{name} must be greater than 0"
+        )));
+    }
+    Ok(value)
+}
+
+fn u32_list(value: &Bound<'_, PyAny>, name: &str) -> PyResult<Vec<u32>> {
+    value
+        .extract::<Vec<i64>>()?
+        .into_iter()
+        .map(|value| {
+            u32::try_from(value).map_err(|_| {
+                PyValueError::new_err(format!(
+                    "{name} must contain values between 0 and {}",
+                    u32::MAX
+                ))
+            })
+        })
+        .collect()
+}
+
+fn optional_i64_bounded_u64(value: &Bound<'_, PyAny>, name: &str) -> PyResult<Option<u64>> {
+    let value: Option<u64> = value.extract()?;
+    if value.is_some_and(|value| value > i64::MAX as u64) {
+        return Err(PyValueError::new_err(format!(
+            "{name} must be at most {}",
+            i64::MAX
+        )));
+    }
+    Ok(value)
+}
+
+fn parse_compaction_options(options: Option<&Bound<'_, PyDict>>) -> PyResult<CompactionOptions> {
+    let mut parsed = CompactionOptions::default();
+    let Some(options) = options else {
+        return Ok(parsed);
+    };
+
+    for (key, value) in options.iter() {
+        let key: String = key.extract()?;
+        match key.as_str() {
+            "target_rows_per_fragment" => {
+                parsed.target_rows_per_fragment = positive_u32(&value, &key)?
+            }
+            "max_rows_per_group" => parsed.max_rows_per_group = positive_u32(&value, &key)?,
+            "max_bytes_per_file" => parsed.max_bytes_per_file = value.extract()?,
+            "materialize_deletions" => parsed.materialize_deletions = value.extract()?,
+            "materialize_deletions_threshold" => {
+                parsed.materialize_deletions_threshold = value.extract()?
+            }
+            "num_threads" => parsed.num_threads = optional_positive_usize(&value, &key)?,
+            "batch_size" => parsed.batch_size = optional_positive_u32(&value, &key)?,
+            "io_buffer_size" => parsed.io_buffer_size = optional_i64_bounded_u64(&value, &key)?,
+            "defer_index_remap" => parsed.defer_index_remap = value.extract()?,
+            "index_remap_mode" => {
+                let mode: String = value.extract()?;
+                parsed.index_remap_mode = IndexRemapMode::try_from(mode.as_str())
+                    .map_err(|err| PyValueError::new_err(err.to_string()))?;
+            }
+            "compaction_mode" => {
+                let mode: Option<String> = value.extract()?;
+                parsed.compaction_mode = mode
+                    .map(|mode| {
+                        CompactionMode::try_from(mode.as_str())
+                            .map_err(|err| PyValueError::new_err(err.to_string()))
+                    })
+                    .transpose()?;
+            }
+            "binary_copy_read_batch_bytes" => {
+                parsed.binary_copy_read_batch_bytes = value.extract()?
+            }
+            "max_source_fragments" => parsed.max_source_fragments = value.extract()?,
+            "max_source_rows" => parsed.max_source_rows = optional_positive_usize(&value, &key)?,
+            "max_source_bytes" => parsed.max_source_bytes = optional_positive_u64(&value, &key)?,
+            "excluded_fragment_ids" => parsed.excluded_fragment_ids = u32_list(&value, &key)?,
+            "max_overlays_per_fragment" => parsed.max_overlays_per_fragment = value.extract()?,
+            _ => {
+                return Err(PyValueError::new_err(format!(
+                    "Invalid compaction option: {key}"
+                )));
+            }
+        }
+    }
+
+    Ok(parsed)
+}
+
 /// Statistics about a compaction operation.
 #[pyclass(get_all, from_py_object)]
 #[derive(Clone, Debug)]
@@ -1340,13 +1463,15 @@ impl Table {
    }

    /// Optimize the on-disk data by compacting and pruning old data, for better performance.
-    #[pyo3(signature = (cleanup_since_ms=None, delete_unverified=None))]
-    pub fn optimize(
-        self_: PyRef<'_, Self>,
+    #[pyo3(signature = (cleanup_since_ms=None, delete_unverified=None, compaction_options=None))]
+    pub fn optimize<'py>(
+        self_: PyRef<'py, Self>,
        cleanup_since_ms: Option<u64>,
        delete_unverified: Option<bool>,
-    ) -> PyResult<Bound<'_, PyAny>> {
+        compaction_options: Option<&Bound<'py, PyDict>>,
+    ) -> PyResult<Bound<'py, PyAny>> {
        let inner = self_.inner_ref()?.clone();
+        let compaction_options = parse_compaction_options(compaction_options)?;
        let older_than = if let Some(ms) = cleanup_since_ms {
            if ms > i64::MAX as u64 {
                return Err(PyValueError::new_err(format!(
@@ -1362,7 +1487,7 @@ impl Table {
        future_into_py(self_.py(), async move {
            let compaction_stats = inner
                .optimize(OptimizeAction::Compact {
-                    options: lancedb::table::CompactionOptions::default(),
+                    options: compaction_options,
                    remap_options: None,
                })
                .await
@@ -1492,7 +1617,7 @@ impl Table {
        })
    }

-    /// Freeze every table shard's active memtable into an SSTable.
+    /// Seal every bucket's active memtable into L0.
    pub fn flush_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
        let inner = self_.inner_ref()?.clone();
        future_into_py(
@@ -1501,7 +1626,7 @@ impl Table {
        )
    }

-    /// Trigger a background SSTable compaction pass per table shard. Returns once the
+    /// Trigger a background L0 → base pass per bucket. Returns once the
    /// passes are dispatched, not once they finish — watch `get_lsm_stats`.
    pub fn compact_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
        let inner = self_.inner_ref()?.clone();
@@ -1511,15 +1636,15 @@ impl Table {
    }

    /// Live LSM state, or `None` when the LSM write path is not enabled.
-    #[pyo3(signature = (include_sstable_rows=false))]
+    #[pyo3(signature = (include_generation_rows=false))]
    pub fn get_lsm_stats(
        self_: PyRef<'_, Self>,
-        include_sstable_rows: bool,
+        include_generation_rows: bool,
    ) -> PyResult<Bound<'_, PyAny>> {
        let inner = self_.inner_ref()?.clone();
        future_into_py(self_.py(), async move {
            let stats = inner
-                .get_lsm_stats(include_sstable_rows)
+                .get_lsm_stats(include_generation_rows)
                .await
                .infer_error()?;
            Python::attach(|py| stats.map(|s| lsm_stats_to_py(py, &s)).transpose())
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.38.0-beta.11"
+version = "0.38.0-beta.12"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -13,7 +13,7 @@ use lance::dataset::{ReadParams, WriteMode, builder::DatasetBuilder};
 use lance::io::{ObjectStore, ObjectStoreParams, WrappingObjectStore};
 use lance_datafusion::utils::StreamingWriteSource;
 use lance_file::version::LanceFileVersion;
-use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
+use lance_io::object_store::{ReadDirOptions, StorageOptionsAccessor, StorageOptionsProvider};
 use lance_table::io::commit::commit_handler_from_url;
 use object_store::local::LocalFileSystem;
 use snafu::ResultExt;
@@ -281,6 +281,22 @@ impl std::fmt::Display for ListingDatabase {
 }

 const LANCE_EXTENSION: &str = "lance";
+
+/// The table a listed child of the database names, or `None` if the child is not a table.
+///
+/// A table is the directory `<name>.lance`; a loose file or any other directory under the
+/// database prefix belongs to something else. `dir_suffix` is `.lance`, built once by the
+/// caller rather than per child.
+/// The table a listed child directory holds, or `None` if it is not a table at all.
+///
+/// Only directories are considered, so a loose object named like a table is not one.
+fn table_name(location: &object_store::path::Path, dir_suffix: &str) -> Option<String> {
+    location
+        .filename()?
+        .strip_suffix(dir_suffix)
+        .map(String::from)
+        .filter(|name| !name.is_empty())
+}
 const ENGINE: &str = "engine";
 const MIRRORED_STORE: &str = "mirroredStore";

@@ -944,51 +960,72 @@ impl Database for ListingDatabase {
        Ok(f)
    }

+    /// List the tables in the database, a page at a time.
+    ///
+    /// The page_token is opaque, unlike the `start_after` parameter of [`Self::table_names()`].
+    ///
+    /// When there are no more results, the returned page_token will be None.
+    ///
+    /// `limit` is the maximum number of tables to return in the response. But it is possible
+    /// for the response to contain fewer than `limit` tables, even when there are more tables
+    /// to return. Clients should check the returned page_token to determine if there are
+    /// more results, rather than relying on the number of tables returned.
+    ///
+    /// The order that results are returned in not guaranteed to be stable across calls,
+    /// so clients should not rely on it.
    async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
        if request.id.as_ref().map(|v| !v.is_empty()).unwrap_or(false) {
            return self.namespace_database().list_tables(request).await;
        }
-        let mut f = self
-            .object_store
-            .read_dir(self.base_path.clone())
-            .await?
-            .iter()
-            .map(Path::new)
-            .filter(|path| {
-                let is_lance = path
-                    .extension()
-                    .and_then(|e| e.to_str())
-                    .map(|e| e == LANCE_EXTENSION);
-                is_lance.unwrap_or(false)
-            })
-            .filter_map(|p| p.file_stem().and_then(|s| s.to_str().map(String::from)))
-            .collect::<Vec<String>>();
-        f.sort();
+        let limit = request.limit.map(|limit| limit.max(0) as usize);
+        let dir_suffix = format!(".{LANCE_EXTENSION}");
+        let mut tables = Vec::new();
+        let mut page_token = request.page_token.filter(|token| !token.is_empty());

-        // Handle pagination with page_token
-        if let Some(ref page_token) = request.page_token {
-            let index = f
-                .iter()
-                .position(|name| name.as_str() > page_token.as_str())
-                .unwrap_or(f.len());
-            f.drain(0..index);
+        // A page of nothing: the store rejects a limit of zero, and no table was handed over
+        // for a token to resume after.
+        if limit == Some(0) {
+            return Ok(ListTablesResponse {
+                context: None,
+                tables,
+                page_token: None,
+            });
        }

-        // Determine if there's a next page. The token is the last name of this page,
-        // not the first of the next one: the next page resumes strictly after the
-        // token, so naming the next page's first entry would skip it.
-        let next_page_token = match request.limit {
-            Some(limit) if f.len() > limit as usize => {
-                f.truncate(limit as usize);
-                f.last().cloned()
+        loop {
+            // Ask only for what the page still has room for, so a database holding more
+            // than one page costs one request per page rather than one per table.
+            let listing = self
+                .object_store
+                .read_dir_page(
+                    self.base_path.clone(),
+                    ReadDirOptions {
+                        page_token: page_token.take(),
+                        limit: limit.map(|limit| limit - tables.len()),
+                    },
+                )
+                .await?;
+            page_token = listing.page_token;
+            // Only child directories can be tables, and the store already separates them
+            // out, so the objects in the page are not looked at.
+            tables.extend(
+                listing
+                    .result
+                    .common_prefixes
+                    .iter()
+                    .filter_map(|location| table_name(location, &dir_suffix)),
+            );
+            // Children that are not tables leave the page short of the limit, so keep
+            // going until the page is full or the database runs out.
+            if page_token.is_none() || limit.is_none_or(|limit| tables.len() >= limit) {
+                break;
            }
-            _ => None,
-        };
+        }

        Ok(ListTablesResponse {
            context: None,
-            tables: f,
-            page_token: next_page_token,
+            tables,
+            page_token,
        })
    }

@@ -1484,6 +1521,182 @@ mod tests {
    use tokio::sync::Barrier;
    use tokio::time::timeout;

+    async fn create_tables(db: &ListingDatabase, names: &[&str]) {
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        for name in names {
+            db.create_table(CreateTableRequest {
+                name: name.to_string(),
+                namespace_path: vec![],
+                data: Box::new(RecordBatch::new_empty(schema.clone())) as Box<dyn Scannable>,
+                mode: CreateTableMode::Create,
+                write_options: Default::default(),
+                location: None,
+                namespace_client: None,
+            })
+            .await
+            .unwrap();
+        }
+    }
+
+    /// Every table in the database, taken `limit` at a time, which is how a caller walks a
+    /// listing: the token ends the walk, never a short page.
+    async fn walk(db: &ListingDatabase, limit: Option<i32>) -> Vec<String> {
+        let mut seen = Vec::new();
+        let mut page_token = None;
+        loop {
+            let page = db
+                .list_tables(ListTablesRequest {
+                    limit,
+                    page_token,
+                    ..Default::default()
+                })
+                .await
+                .unwrap();
+            seen.extend(page.tables);
+            page_token = page.page_token;
+            if page_token.is_none() {
+                return seen;
+            }
+            assert!(
+                seen.len() < 100,
+                "the walk is serving tables more than once"
+            );
+        }
+    }
+
+    /// Paging with the returned token has to visit every table exactly once, whatever the
+    /// page size, with nothing lost or repeated at a boundary.
+    #[rstest::rstest]
+    #[tokio::test]
+    async fn test_list_tables_pages_over_every_table_once(#[values(1, 2, 3, 5, 10)] limit: i32) {
+        let (_tempdir, db) = setup_database().await;
+        create_tables(&db, &["a", "b", "c", "d", "e"]).await;
+
+        assert_eq!(walk(&db, Some(limit)).await, vec!["a", "b", "c", "d", "e"]);
+    }
+
+    /// The token is opaque: it is whatever resumes the store the database sits on, not a
+    /// table name. Callers hand it back and nothing else.
+    ///
+    /// Nothing validates a token, so one invented by a caller is read as a position rather
+    /// than refused — which is why the token has to come back from a previous page.
+    #[tokio::test]
+    async fn test_the_page_token_is_not_a_table_name() {
+        let (_tempdir, db) = setup_database().await;
+        create_tables(&db, &["a", "b", "c"]).await;
+
+        let page = db
+            .list_tables(ListTablesRequest {
+                limit: Some(1),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(page.tables, vec!["a"]);
+        let token = page.page_token.expect("two tables are still to come");
+        assert_ne!(token, "a");
+
+        // Handing it back is the only thing a caller does with it, and it resumes.
+        let rest = db
+            .list_tables(ListTablesRequest {
+                page_token: Some(token),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        assert_eq!(rest.tables, vec!["b", "c"]);
+    }
+
+    /// A limit the listing does not fill leaves no token behind, so a caller paging by token
+    /// stops without asking for an empty page.
+    #[tokio::test]
+    async fn test_a_listing_that_runs_out_has_no_token() {
+        let (_tempdir, db) = setup_database().await;
+        create_tables(&db, &["a", "b"]).await;
+
+        let page = db
+            .list_tables(ListTablesRequest {
+                limit: Some(10),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(page.tables, vec!["a", "b"]);
+        assert_eq!(page.page_token, None);
+    }
+
+    /// An empty page token means "from the start", which is how a client looping on a token
+    /// spells its first request.
+    #[tokio::test]
+    async fn test_an_empty_page_token_lists_from_the_start() {
+        let (_tempdir, db) = setup_database().await;
+        create_tables(&db, &["a", "b"]).await;
+
+        let page = db
+            .list_tables(ListTablesRequest {
+                page_token: Some(String::new()),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(page.tables, vec!["a", "b"]);
+    }
+
+    /// Listing follows the order the object store lists directories in, so a name that
+    /// extends another comes first: the `-` of `users-archive.lance` sorts below the `.` of
+    /// `users.lance`. Pagination pushes its cursor into the list request, so it cannot report
+    /// an order other than the one it resumes in.
+    #[tokio::test]
+    async fn test_listing_order_follows_the_store_not_the_table_name() {
+        let (_tempdir, db) = setup_database().await;
+        create_tables(&db, &["users", "users-archive", "users.old"]).await;
+
+        assert_eq!(
+            walk(&db, None).await,
+            vec!["users-archive", "users", "users.old"]
+        );
+        // And paging reports the same order, so a walk sees each table once.
+        assert_eq!(
+            walk(&db, Some(1)).await,
+            vec!["users-archive", "users", "users.old"]
+        );
+    }
+
+    /// Only directories named `<name>.lance` are tables; loose files and other directories
+    /// under the database prefix are not. A page spent on them is filled from the next one,
+    /// so a page holding only non-tables does not read as an empty database.
+    #[tokio::test]
+    async fn test_listing_ignores_non_table_children() {
+        let (tempdir, db) = setup_database().await;
+        create_tables(&db, &["real"]).await;
+        std::fs::write(tempdir.path().join("aaa-loose.lance"), b"not a table").unwrap();
+        create_dir_all(tempdir.path().join("aaa-scratch")).unwrap();
+
+        let page = db
+            .list_tables(ListTablesRequest {
+                limit: Some(1),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(page.tables, vec!["real"]);
+    }
+
+    #[tokio::test]
+    async fn listing_ignores_empty_table_name() {
+        let (tempdir, db) = setup_database().await;
+        create_dir_all(tempdir.path().join(".lance")).unwrap();
+        let page = db.list_tables(ListTablesRequest::default()).await.unwrap();
+        assert!(
+            page.tables.is_empty(),
+            "invalid empty table name was listed"
+        );
+    }
+
    async fn setup_database() -> (tempfile::TempDir, ListingDatabase) {
        let tempdir = tempdir().unwrap();
        let uri = tempdir.path().to_str().unwrap();
@@ -878,7 +878,7 @@ pub struct QueryRequest {
    /// [`crate::Table::set_lsm_write_spec`]) is routed through the LSM scanner so
    /// it also sees data written via the `merge_insert` LSM path that has not yet
    /// been compacted into the base table — the active and frozen in-memory
-    /// memtables and the SSTables, deduplicated by primary key
+    /// memtables and the flushed (L0) generations, deduplicated by primary key
    /// against the base table (newest generation wins); a table without a spec
    /// reads the base table.
    ///
@@ -2951,13 +2951,13 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        Ok(())
    }

-    async fn get_lsm_stats(&self, include_sstable_rows: bool) -> Result<Option<LsmStats>> {
+    async fn get_lsm_stats(&self, include_generation_rows: bool) -> Result<Option<LsmStats>> {
        // Read-semantics POST, like `get_lsm_write_spec`.
        let request = self
            .client
            .post(&format!("/v1/table/{}/get_lsm_stats/", self.identifier))
            .json(&serde_json::json!({
-                "include_sstable_rows": include_sstable_rows,
+                "include_generation_rows": include_generation_rows,
            }));
        let (request_id, response) = self.send_lsm_route(request).await?;
        let body = response.text().await.err_to_http(request_id.clone())?;
@@ -3180,8 +3180,8 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
            self.schema().await?.as_ref(),
            "schema evolution",
        )?;
-        // The server plans the declaration: expression validation, type
-        // inference and the persisted binding all happen there.
+        // The server plans the declaration against its table schema, including
+        // Blob v2 semantics inherited by a direct field projection.
        let entries = columns
            .iter()
            .map(
@@ -7388,8 +7388,8 @@ mod tests {
        assert_eq!(result.version, if old_server { 0 } else { 43 });
    }

-    /// A declaration is sent as `{name, computed}` entries for the server to
-    /// plan; the client never types the expression itself.
+    /// A declaration is sent as `{name, computed}` for the server to plan; the
+    /// client never types the expression itself.
    #[tokio::test]
    async fn test_add_computed_columns_sends_the_expression() {
        let table = Table::new_with_handler("my_table", |request| match request.url().path() {
@@ -8260,7 +8260,7 @@ mod tests {
        http::Response::builder().status(200).body(body).unwrap()
    }

-    /// A flush landing in an empty SSTable tier finishes on the opening stats read
+    /// A flush landing in an empty L0 finishes on the opening stats read
    /// alone. Asserting zero compacts is the point: "it returned Ok" is also
    /// true of a loop that ran a pointless pass.
    #[tokio::test(start_paused = true)]
@@ -8314,7 +8314,7 @@ mod tests {
    }

    /// Generations created *during* the checkpoint are not waited on, which
-    /// is what lets the loop terminate on a table taking writes where "the SSTable tier is
+    /// is what lets the loop terminate on a table taking writes where "L0 is
    /// empty" never becomes true.
    #[tokio::test(start_paused = true)]
    async fn test_checkpoint_ignores_generations_created_while_it_runs() {
@@ -8593,7 +8593,7 @@ mod tests {
    }

    /// WAL off ⇒ `None`; WAL on ⇒ a fully populated `Some` with no field
-    /// defaulting to a zero it did not measure. `include_sstable_rows`
+    /// defaulting to a zero it did not measure. `include_generation_rows`
    /// rides in the body and is off unless asked for.
    #[tokio::test]
    async fn test_get_lsm_stats_round_trip() {
@@ -8602,7 +8602,7 @@ mod tests {
            let body = request.body().unwrap().as_bytes().unwrap();
            let body: serde_json::Value = serde_json::from_slice(body).unwrap();
            assert_eq!(
-                body["include_sstable_rows"], true,
+                body["include_generation_rows"], true,
                "the flag must reach the server, not be silently dropped"
            );
            let response = serde_json::json!({
@@ -102,8 +102,10 @@ use futures::future::join_all;
 pub use lance::dataset::refs::{BranchContents, Ref, TagContents, Tags as LanceTags};
 pub use lance::dataset::scanner::DatasetRecordBatchStream;
 pub use lance_index::optimize::OptimizeOptions;
-pub use lsm_stats::{LsmStats, MemtableStats, SsTableStats, TableShardStats};
-pub use optimize::{CompactionOptions, OptimizeAction, OptimizeStats};
+pub use lsm_stats::{BucketStats, GenerationStats, LsmStats, MemtableStats};
+pub use optimize::{
+    CompactionMode, CompactionOptions, IndexRemapMode, OptimizeAction, OptimizeStats,
+};
 pub use refresh::RefreshColumnResult;
 pub use schema_evolution::{
    AddColumnsResult, AlterColumnsResult, DropColumnsResult, FieldMetadataUpdate,
@@ -673,7 +675,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
            message: "get_lsm_write_spec is not supported on this table type".into(),
        })
    }
-    /// Freeze every table shard's active memtable into an SSTable.
+    /// Seal every bucket's active memtable into L0.
    ///
    /// The default implementation returns `NotSupported`.
    async fn flush_lsm(&self) -> Result<()> {
@@ -681,7 +683,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
            message: "flush_lsm is not supported on this table type".into(),
        })
    }
-    /// Trigger a background SSTable compaction pass per table shard.
+    /// Trigger a background L0 → base compaction pass per bucket.
    ///
    /// The default implementation returns `NotSupported`.
    async fn compact_lsm(&self) -> Result<()> {
@@ -693,7 +695,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
    /// enabled for this table.
    ///
    /// The default implementation returns `NotSupported`.
-    async fn get_lsm_stats(&self, _include_sstable_rows: bool) -> Result<Option<LsmStats>> {
+    async fn get_lsm_stats(&self, _include_generation_rows: bool) -> Result<Option<LsmStats>> {
        Err(Error::NotSupported {
            message: "get_lsm_stats is not supported on this table type".into(),
        })
@@ -750,8 +752,8 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
    /// Declare computed columns, each defined by a SQL expression.
    ///
    /// Where the declaration is planned depends on the backend: a local table
-    /// validates and types the expression itself, a remote one sends the text
-    /// for the server to plan.
+    /// validates and types the expression itself, while a remote one sends the
+    /// expression for the server to plan.
    async fn add_computed_columns(
        &self,
        _columns: &[(String, String)],
@@ -1897,7 +1899,7 @@ impl Table {

    /// Converge this table's LSM write path into its base table.
    ///
-    /// One `flush` to freeze every memtable into an SSTable, then compaction triggers
+    /// One `flush` to seal every memtable into L0, then compaction triggers
    /// until every generation that existed at that moment has reached base.
    /// The loop runs client-side, reading progress from `get_lsm_stats`, so
    /// there is no held socket and nothing to reconcile if you drop this
@@ -1932,10 +1934,10 @@ impl Table {
        checkpoint::checkpoint_lsm(self).await
    }

-    /// Freeze every table shard's active memtable into an SSTable without touching the
+    /// Seal every bucket's active memtable into L0 without touching the
    /// base table.
    ///
-    /// Independently useful: flushing makes memtable rows readable from an SSTable at
+    /// Independently useful: flushing makes memtable rows readable from L0 at
    /// a lower per-query cost. On a node that has not claimed this table it
    /// claims it and replays the WAL log first — reporting "nothing to flush"
    /// without replaying would lie about durable data.
@@ -1943,7 +1945,7 @@ impl Table {
        self.inner.flush_lsm().await
    }

-    /// Run one bounded SSTable compaction pass per table shard, reporting what
+    /// Run one bounded L0 → base compaction pass per bucket, reporting what
    /// it merged and what is left.
    ///
    /// One pass, not convergence: that bounds each request's cost and gives a
@@ -1959,7 +1961,7 @@ impl Table {
    /// state, though on a node that has not claimed this table it claims it,
    /// exactly as a read would.
    ///
-    /// `include_sstable_rows` reports a row count per SSTable. Off by
+    /// `include_generation_rows` reports a row count per L0 generation. Off by
    /// default: each count opens an uncached Lance dataset, and
    /// `checkpoint_lsm` polls this needing only generation numbers.
    ///
@@ -1970,8 +1972,8 @@ impl Table {
    ///
    /// Do not build a checkpoint's termination on this: the completion
    /// predicate lives in the `flush` and `compact` responses.
-    pub async fn get_lsm_stats(&self, include_sstable_rows: bool) -> Result<Option<LsmStats>> {
-        self.inner.get_lsm_stats(include_sstable_rows).await
+    pub async fn get_lsm_stats(&self, include_generation_rows: bool) -> Result<Option<LsmStats>> {
+        self.inner.get_lsm_stats(include_generation_rows).await
    }

    /// Drain and close any cached MemWAL shard writers held for this table.
@@ -4,7 +4,7 @@
 //! Converging a table's LSM write path into its base table.
 //!
 //! `checkpoint_lsm` seals once, then triggers compaction and watches
-//! generation numbers until the SSTables that existed at the start are gone.
+//! generation numbers until the L0 that existed at the start is gone.
 //!
 //! The loop runs in the client, not the server: `compact_lsm` dispatches a
 //! pass and returns, so nothing holds a socket and a client can vanish
@@ -150,7 +150,7 @@ where
 }

 /// Drive [`Table::checkpoint_lsm`]: seal once, fix the target watermark
-/// from the resulting SSTables, then trigger and poll until they drain.
+/// from the resulting L0, then trigger and poll until it drains.
 pub(crate) async fn checkpoint_lsm(table: &Table) -> Result<()> {
    for reissue in 0..=MAX_REISSUES {
        // The seal turns everything written before this call into a
@@ -177,9 +177,9 @@ pub(crate) async fn checkpoint_lsm(table: &Table) -> Result<()> {
            return Ok(());
        };
        let targets: HashMap<String, u64> = stats
-            .table_shards
+            .buckets
            .iter()
-            .filter_map(|b| Some((b.shard_id.clone(), b.newest_sstable_generation()?)))
+            .filter_map(|b| Some((b.shard_id.clone(), b.newest_generation()?)))
            .collect();
        if targets.is_empty() {
            return Ok(());
@@ -226,11 +226,11 @@ async fn drain_to_targets(
        // with nothing outstanding are skipped, not counted as idle.
        let mut outstanding = 0;
        let mut all_compacting = true;
-        for b in &stats.table_shards {
+        for b in &stats.buckets {
            let Some(target) = targets.get(&b.shard_id) else {
                continue;
            };
-            let n = b.outstanding_sstables(*target);
+            let n = b.outstanding_generations(*target);
            if n > 0 {
                outstanding += n;
                all_compacting &= b.compacting;
@@ -9,29 +9,35 @@
 //! refresh fills the rows.
 //!
 //! The rule is tagged by kind ([`ComputedColumnKind`]) because kinds differ in
-//! where the column's type and inputs come from. A SQL expression is
-//! self-describing -- both are derived from the expression, so a caller writes
-//! neither -- while a kind resolved through a registry cannot be typed without
-//! consulting it. Registered Functions use an exact remote version plus a
-//! schema-level Function binding; unknown newer kinds remain readable and fail
-//! closed before mutation.
+//! where the column's type and inputs come from. A SQL expression determines
+//! its inputs and physical result type. A direct projection of a Blob v2 field
+//! also inherits that field's semantic type while execution continues to use
+//! `LargeBinary`. A kind resolved through a registry cannot be typed without
+//! consulting it.
+//! Registered Functions use an exact remote version plus a schema-level
+//! Function binding; unknown newer kinds remain readable and fail closed
+//! before mutation.
 //!
 //! [`computed_columns`] and [`computed_column_from_field`] read declarations
 //! back off a schema.

-use std::collections::{BTreeSet, HashMap};
+use std::collections::{BTreeSet, HashMap, HashSet};
 use std::sync::Arc;

-use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema, SchemaRef};
-use datafusion_common::tree_node::TreeNode;
+use arrow_schema::{DataType, Field as ArrowField, Fields, Schema as ArrowSchema, SchemaRef};
+use datafusion_common::{ScalarValue, tree_node::TreeNode};
+use datafusion_expr::Expr;
 use datafusion_physical_plan::PhysicalExpr;
 use lance::dataset::NewColumnTransform;
+use lance_arrow::FieldExt;
+use lance_core::datatypes::{BLOB_V2_DESC_FIELD, format_field_path_minimal, parse_field_path};
 use lance_datafusion::planner::Planner;
 use lance_namespace::models::{JsonArrowDataType, JsonArrowField, JsonArrowSchema};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;

 use crate::function::{FunctionApplication, FunctionBinding};
+use crate::utils::resolve_arrow_field_path;
 use crate::{Error, Result};

 /// Field metadata key marking a column as computed. The value is `"true"`.
@@ -1106,15 +1112,20 @@ pub(crate) fn ensure_no_foreign_declarations<'a>(
    fields: impl IntoIterator<Item = &'a Arc<ArrowField>>,
 ) -> Result<()> {
    for field in fields {
-        if field.metadata().keys().any(|k| is_declaration_key(k)) {
-            return Err(Error::InvalidInput {
-                message: format!(
-                    "field '{}' carries computed-column metadata; declare computed columns \
-                     with add_columns().computed()",
-                    field.name()
-                ),
-            });
-        }
+        ensure_no_foreign_declaration(field)?;
+    }
+    Ok(())
+}
+
+fn ensure_no_foreign_declaration(field: &ArrowField) -> Result<()> {
+    if field.metadata().keys().any(|k| is_declaration_key(k)) {
+        return Err(Error::InvalidInput {
+            message: format!(
+                "field '{}' carries computed-column metadata; declare computed columns \
+                 with add_columns().computed()",
+                field.name()
+            ),
+        });
    }
    Ok(())
 }
@@ -1162,15 +1173,154 @@ pub(crate) struct BoundExpression {
    /// The columns the expression names, as written; nested inputs keep
    /// their dotted path.
    pub inputs: Vec<String>,
-    /// The top-level columns evaluation reads, in [`Self::read_schema`]
-    /// order. A nested input appears through its root.
+    /// The top-level columns evaluation reads, in physical-expression order.
+    /// A nested input appears through its root.
    pub roots: Vec<String>,
-    /// The projected schema evaluation runs against.
-    pub read_schema: SchemaRef,
    /// The compiled expression.
    pub physical: Arc<dyn PhysicalExpr>,
    /// The type the expression yields.
    pub data_type: DataType,
+    /// Blob v2 leaves the scan must materialize as `LargeBinary`.
+    pub blob_paths: Vec<String>,
+    /// A directly projected Blob v2 field whose semantics the output inherits.
+    projected_blob_field: Option<ArrowField>,
+}
+
+fn is_direct_field_projection(expr: &Expr) -> bool {
+    match expr {
+        Expr::Column(_) => true,
+        Expr::ScalarFunction(function)
+            if function.name() == "get_field" && function.args.len() == 2 =>
+        {
+            is_direct_field_projection(&function.args[0])
+                && matches!(
+                    &function.args[1],
+                    Expr::Literal(ScalarValue::Utf8(Some(_)), _)
+                )
+        }
+        _ => false,
+    }
+}
+
+fn projected_blob_field(schema: &ArrowSchema, expr: &Expr) -> Result<Option<ArrowField>> {
+    if !is_direct_field_projection(expr) {
+        return Ok(None);
+    }
+    let paths = Planner::column_names_in_expr(expr);
+    let [path] = paths.as_slice() else {
+        return Ok(None);
+    };
+    let (_, field) = resolve_arrow_field_path(schema, path)?;
+    Ok(field.is_blob_v2().then_some(field))
+}
+
+fn collect_blob_paths(field: &ArrowField, parent: &[String], paths: &mut Vec<Vec<String>>) {
+    let mut path = parent.to_vec();
+    path.push(field.name().clone());
+    if field.is_blob_v2() {
+        paths.push(path);
+        return;
+    }
+    match field.data_type() {
+        DataType::Struct(children) => {
+            for child in children {
+                collect_blob_paths(child, &path, paths);
+            }
+        }
+        DataType::List(child)
+        | DataType::LargeList(child)
+        | DataType::FixedSizeList(child, _)
+        | DataType::Map(child, _) => collect_blob_paths(child, &path, paths),
+        _ => {}
+    }
+}
+
+fn schema_blob_paths(schema: &ArrowSchema) -> Vec<Vec<String>> {
+    let mut paths = Vec::new();
+    for field in schema.fields() {
+        collect_blob_paths(field, &[], &mut paths);
+    }
+    paths
+}
+
+fn transform_blob_field(
+    field: &ArrowField,
+    parent: &[String],
+    materialized: &HashSet<Vec<String>>,
+) -> ArrowField {
+    let mut path = parent.to_vec();
+    path.push(field.name().clone());
+    if field.is_blob_v2() {
+        if materialized.contains(&path) {
+            return ArrowField::new(field.name(), DataType::LargeBinary, field.is_nullable());
+        }
+        return ArrowField::new(
+            field.name(),
+            BLOB_V2_DESC_FIELD.data_type().clone(),
+            field.is_nullable(),
+        )
+        .with_metadata(BLOB_V2_DESC_FIELD.metadata().clone());
+    }
+
+    let data_type = match field.data_type() {
+        DataType::Struct(children) => DataType::Struct(
+            children
+                .iter()
+                .map(|child| Arc::new(transform_blob_field(child, &path, materialized)))
+                .collect(),
+        ),
+        DataType::List(child) => {
+            DataType::List(Arc::new(transform_blob_field(child, &path, materialized)))
+        }
+        DataType::LargeList(child) => {
+            DataType::LargeList(Arc::new(transform_blob_field(child, &path, materialized)))
+        }
+        DataType::FixedSizeList(child, size) => DataType::FixedSizeList(
+            Arc::new(transform_blob_field(child, &path, materialized)),
+            *size,
+        ),
+        DataType::Map(child, sorted) => DataType::Map(
+            Arc::new(transform_blob_field(child, &path, materialized)),
+            *sorted,
+        ),
+        _ => return field.clone(),
+    };
+    ArrowField::new(field.name(), data_type, field.is_nullable())
+        .with_metadata(field.metadata().clone())
+}
+
+fn blob_runtime_schema(schema: &ArrowSchema, materialized: &HashSet<Vec<String>>) -> SchemaRef {
+    Arc::new(ArrowSchema::new_with_metadata(
+        schema
+            .fields()
+            .iter()
+            .map(|field| Arc::new(transform_blob_field(field, &[], materialized)))
+            .collect::<Fields>(),
+        schema.metadata().clone(),
+    ))
+}
+
+fn referenced_blob_paths(schema: &ArrowSchema, inputs: &[String]) -> Result<Vec<Vec<String>>> {
+    let input_paths = inputs
+        .iter()
+        .map(|input| {
+            parse_field_path(input).map_err(|error| Error::InvalidInput {
+                message: format!("invalid computed-column input path '{input}': {error}"),
+            })
+        })
+        .collect::<Result<Vec<_>>>()?;
+    Ok(schema_blob_paths(schema)
+        .into_iter()
+        .filter(|blob_path| {
+            input_paths.iter().any(|input_path| {
+                input_path.len() <= blob_path.len()
+                    && input_path
+                        .iter()
+                        .zip(blob_path)
+                        .all(|(input, blob)| input == blob)
+            })
+        })
+        .collect())
 }

 /// Parse, resolve and compile `expression` against `schema`.
@@ -1185,10 +1335,18 @@ pub(crate) fn bind(schema: SchemaRef, column: &str, expression: &str) -> Result<
        message,
    };

-    let planner = Planner::new(schema.clone());
+    // Blob v2 is a semantic type whose runtime expression ABI is
+    // `LargeBinary`. Parse against that ABI first so a direct Blob reference
+    // is not mistaken for its storage descriptor struct.
+    let all_blob_paths = schema_blob_paths(schema.as_ref())
+        .into_iter()
+        .collect::<HashSet<_>>();
+    let parsing_schema = blob_runtime_schema(schema.as_ref(), &all_blob_paths);
+    let planner = Planner::new(parsing_schema);
    let parsed = planner
        .parse_expr(expression)
        .map_err(|e| invalid(e.to_string()))?;
+    let projected_blob_field = projected_blob_field(schema.as_ref(), &parsed)?;

    // A declaration is evaluated more than once -- staging and writing are
    // separate passes, and a refresh years later replays the same text -- so
@@ -1218,13 +1376,19 @@ pub(crate) fn bind(schema: SchemaRef, column: &str, expression: &str) -> Result<
    inputs.sort();
    inputs.dedup();

+    let blob_paths = referenced_blob_paths(schema.as_ref(), &inputs)?;
+    let runtime_schema = blob_runtime_schema(
+        schema.as_ref(),
+        &blob_paths.iter().cloned().collect::<HashSet<_>>(),
+    );
+
    // A nested input is recorded by its path but read through its root
    // column; Schema::index_of resolves top-level names only. Resolved here
    // rather than left to the planner so an unknown column names itself in
    // the error instead of surfacing as a plan failure.
    let mut indices = Vec::with_capacity(inputs.len());
    for input in &inputs {
-        let index = schema
+        let index = runtime_schema
            .index_of(root(input))
            .map_err(|_| invalid(format!("unknown column '{input}'")))?;
        if !indices.contains(&index) {
@@ -1237,7 +1401,7 @@ pub(crate) fn bind(schema: SchemaRef, column: &str, expression: &str) -> Result<
    // compiles the expression has to be built on the projected schema
    // evaluation will actually read.
    let read_schema = Arc::new(
-        schema
+        runtime_schema
            .project(&indices)
            .map_err(|e| invalid(e.to_string()))?,
    );
@@ -1247,7 +1411,8 @@ pub(crate) fn bind(schema: SchemaRef, column: &str, expression: &str) -> Result<
        .map(|field| field.name().clone())
        .collect();

-    let optimized = planner
+    let runtime_planner = Planner::new(runtime_schema);
+    let optimized = runtime_planner
        .optimize_expr(parsed)
        .map_err(|e| invalid(e.to_string()))?;
    let physical = Planner::new(read_schema.clone())
@@ -1260,9 +1425,16 @@ pub(crate) fn bind(schema: SchemaRef, column: &str, expression: &str) -> Result<
    Ok(BoundExpression {
        inputs,
        roots,
-        read_schema,
        physical,
        data_type,
+        blob_paths: blob_paths
+            .iter()
+            .map(|path| {
+                let segments = path.iter().map(String::as_str).collect::<Vec<_>>();
+                format_field_path_minimal(&segments)
+            })
+            .collect(),
+        projected_blob_field,
    })
 }

@@ -1273,35 +1445,91 @@ pub(crate) fn bind(schema: SchemaRef, column: &str, expression: &str) -> Result<
 /// refresh time: that the expression parses, that every column it reads
 /// exists, and that the target name is free. A declaration that survives this
 /// is one a refresh can always act on.
-pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Vec<ArrowField>> {
+///
+/// Each accepted column joins the schema the next one resolves against, so a
+/// batch may declare `a` and then `b = a + 1` in one commit. Refresh order
+/// then matters, and refresh enforces it: `b` is refused while `a` still has
+/// unfilled rows.
+fn plan_declarations(schema: SchemaRef, columns: &[(String, String)]) -> Result<Vec<ArrowField>> {
    if columns.is_empty() {
        return Err(Error::InvalidInput {
            message: "at least one computed column is required".into(),
        });
    }

+    let mut schema = schema;
    let mut fields = Vec::with_capacity(columns.len());
-    let mut declared: Vec<&str> = Vec::with_capacity(columns.len());

    for (name, expression) in columns {
-        if schema.field_with_name(name).is_ok() || declared.contains(&name.as_str()) {
-            return Err(Error::ColumnAlreadyExists { name: name.clone() });
+        if schema.field_with_name(name).is_ok() {
+            return Err(Error::ColumnAlreadyExists {
+                name: name.to_string(),
+            });
        }

        let bound = bind(schema.clone(), name, expression)?;

        // Declared columns start entirely null, so nullability is a property
        // of the declaration rather than of what the expression yields.
-        fields.push(
-            ArrowField::new(name, bound.data_type, true)
-                .with_metadata(computed_column_metadata(expression, &bound.inputs)),
-        );
-        declared.push(name);
+        let computed_metadata = computed_column_metadata(expression, &bound.inputs);
+        let field = match bound.projected_blob_field {
+            Some(source) => {
+                let mut metadata = source.metadata().clone();
+                metadata.retain(|key, _| !is_declaration_key(key));
+                metadata.extend(computed_metadata);
+                source
+                    .with_name(name)
+                    .with_nullable(true)
+                    .with_metadata(metadata)
+            }
+            None => ArrowField::new(name, bound.data_type, true).with_metadata(computed_metadata),
+        };
+        schema = Arc::new(ArrowSchema::new_with_metadata(
+            schema
+                .fields()
+                .iter()
+                .cloned()
+                .chain(std::iter::once(Arc::new(field.clone())))
+                .collect::<Fields>(),
+            schema.metadata().clone(),
+        ));
+        fields.push(field);
    }

    Ok(fields)
 }

+pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Vec<ArrowField>> {
+    plan_declarations(schema, columns)
+}
+
+/// Run the schema-level checks of
+/// [`AddColumnsBuilder::computed`](super::AddColumnsBuilder::computed) against
+/// `schema` without committing: the Function-binding guard and the planning of
+/// every declaration. For callers that stage declarations behind other work
+/// and need those rejections before any of it lands.
+///
+/// Only the schema is consulted. Declaring also refuses a table with an LSM
+/// write spec or retained SSTables; that is table state, checked at commit.
+///
+/// ```
+/// # use std::sync::Arc;
+/// # use arrow_schema::{DataType, Field, Schema};
+/// use lancedb::table::computed_columns::validate_declarations;
+///
+/// let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int32, false)]));
+/// let declarations = vec![
+///     ("a".to_string(), "x + 1".to_string()),
+///     ("b".to_string(), "a * 2".to_string()),
+/// ];
+/// assert!(validate_declarations(schema.clone(), &declarations).is_ok());
+/// assert!(validate_declarations(schema, &[("c".into(), "random()".into())]).is_err());
+/// ```
+pub fn validate_declarations(schema: SchemaRef, columns: &[(String, String)]) -> Result<()> {
+    ensure_no_function_bindings_for_mutation(schema.as_ref(), "schema evolution")?;
+    plan(schema, columns).map(drop)
+}
+
 /// Build the transform that declares `columns` against `schema`.
 ///
 /// An all-null column is how a binding with no values yet is carried into a
@@ -1313,7 +1541,7 @@ pub(crate) fn declare(
    schema: SchemaRef,
    columns: &[(String, String)],
 ) -> Result<NewColumnTransform> {
-    let fields = plan(schema, columns)?;
+    let fields = plan_declarations(schema, columns)?;
    Ok(NewColumnTransform::AllNulls(Arc::new(ArrowSchema::new(
        fields,
    ))))
@@ -1340,6 +1568,22 @@ pub(super) async fn add_foreign_kind(table: &crate::Table, name: &str, kind: &st

 #[cfg(test)]
 mod tests {
+    /// The gate's reproducer: the validator applies the same schema-level
+    /// guard declaring does, so a staging caller is refused before it commits
+    /// anything else.
+    #[test]
+    fn test_validate_declarations_matches_schema_admission_barriers() {
+        let schema = Arc::new(ArrowSchema::new_with_metadata(
+            vec![ArrowField::new("x", DataType::Int32, true)],
+            HashMap::from([(
+                FUNCTION_BINDINGS_META_KEY.to_string(),
+                "not valid binding metadata".to_string(),
+            )]),
+        ));
+        let declarations = vec![("a".to_string(), "x + 1".to_string())];
+        assert!(super::validate_declarations(schema, &declarations).is_err());
+    }
+
    #[test]
    fn output_arrow_type_grammar_matches_the_shared_golden() {
        let golden: serde_json::Value = serde_json::from_str(include_str!(
@@ -1423,6 +1667,44 @@ mod tests {
        );
    }

+    #[test]
+    fn test_direct_blob_projection_inherits_semantics() {
+        let schema = Arc::new(ArrowSchema::new(vec![crate::blob("image", false)]));
+        let fields = plan(
+            schema,
+            &[
+                ("first".to_string(), "image".to_string()),
+                ("second".to_string(), "first".to_string()),
+            ],
+        )
+        .unwrap();
+
+        for field in &fields {
+            assert!(field.is_blob_v2());
+            assert!(field.is_nullable());
+        }
+        assert_eq!(
+            fields[1]
+                .metadata()
+                .get(EXPRESSION_META_KEY)
+                .map(String::as_str),
+            Some("first")
+        );
+    }
+
+    #[test]
+    fn test_blob_expression_transformation_does_not_inherit_semantics() {
+        let schema = Arc::new(ArrowSchema::new(vec![crate::blob("image", true)]));
+        let fields = plan(
+            schema,
+            &[("payload".to_string(), "coalesce(image, image)".to_string())],
+        )
+        .unwrap();
+
+        assert!(!fields[0].is_blob_v2());
+        assert_eq!(fields[0].data_type(), &DataType::LargeBinary);
+    }
+
    /// The binding reaches the schema only if `AllNulls` carries per-field
    /// metadata through the commit. The whole representation rests on it.
    #[tokio::test]
@@ -1582,6 +1864,40 @@ mod tests {
        assert!(declared(&table).await.is_empty());
    }

+    /// A batch may build on itself: one commit, and the later entry's inputs
+    /// name the earlier one.
+    #[tokio::test]
+    async fn test_a_declaration_may_read_one_declared_before_it() {
+        let table = table_with_ints("chain").await;
+        let before = table.version().await.unwrap();
+        add_computed(
+            &table,
+            &[("a".into(), "x + 1".into()), ("b".into(), "a * 2".into())],
+        )
+        .await
+        .unwrap();
+        assert_eq!(table.version().await.unwrap(), before + 1);
+        let declared = declared(&table).await;
+        assert_eq!(declared[1].name, "b");
+        assert_eq!(declared[1].inputs, vec!["a".to_string()]);
+
+        // Order is the dependency order; reading ahead is still unknown.
+        let err = add_computed(
+            &table,
+            &[("c".into(), "d + 1".into()), ("d".into(), "x + 1".into())],
+        )
+        .await
+        .unwrap_err();
+        assert!(matches!(err, Error::InvalidExpression { column, .. } if column == "c"));
+        assert!(
+            validate_declarations(
+                table.schema().await.unwrap(),
+                &[("e".into(), "random()".into())]
+            )
+            .is_err()
+        );
+    }
+
    /// A column added by an ordinary transform is materialized, not bound, so
    /// it carries no declaration to report.
    #[tokio::test]
@@ -36,6 +36,14 @@ pub(super) fn coerce_blob_expr(
    };

    let input_shape = match input_field.data_type() {
+        DataType::Null => {
+            let expr: Arc<dyn PhysicalExpr> = Arc::new(CastExpr::new(
+                input_expr,
+                table_field.data_type().clone(),
+                None,
+            ));
+            return Ok((expr, table_field.clone()));
+        }
        DataType::Binary | DataType::LargeBinary | DataType::BinaryView => BlobInputShape::Bytes,
        DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => BlobInputShape::String,
        DataType::Struct(children) => {
@@ -155,7 +163,7 @@ mod tests {
    use crate::blob::blob;
    use arrow_array::{
        Array, ArrayRef, BinaryArray, BinaryViewArray, Int32Array, Int64Array, LargeBinaryArray,
-        RecordBatch, StringArray, StringViewArray, StructArray, UInt8Array, UInt64Array,
+        NullArray, RecordBatch, StringArray, StringViewArray, StructArray, UInt8Array, UInt64Array,
    };
    use arrow_schema::Schema;
    use datafusion::prelude::SessionContext;
@@ -279,6 +287,18 @@ mod tests {
        assert_eq!(data.value(0), b"view");
    }

+    #[tokio::test]
+    async fn null_column_coerces_to_all_null_blob_struct() {
+        let batch = batch_with_image(
+            Field::new("image", DataType::Null, true),
+            Arc::new(NullArray::new(2)),
+        );
+        let coerced = coerce(batch, &blob_table_schema()).await;
+        let image = image_struct(&coerced);
+        assert!(image.is_null(0));
+        assert!(image.is_null(1));
+    }
+
    #[tokio::test]
    async fn binary_nulls_stay_null_after_coercion() {
        let batch = batch_with_image(
@@ -1,21 +1,21 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

-//! Live per-table_shard LSM state — the shape [`crate::Table::get_lsm_stats`]
+//! Live per-bucket LSM state — the shape [`crate::Table::get_lsm_stats`]
 //! returns and [`super::checkpoint`] polls.
 //!
-//! Nothing here is derived: sums and differences (total SSTable bytes, WAL lag)
+//! Nothing here is derived: sums and differences (total L0 bytes, WAL lag)
 //! are the caller's to compute. There is no "WAL is off" shape — that case is
 //! `None`, because a struct of zeros would read as measurements.

 use serde::Deserialize;

-/// One SSTable.
+/// One flushed L0 generation.
 #[derive(Debug, Clone, Deserialize)]
-pub struct SsTableStats {
+pub struct GenerationStats {
    pub generation: u64,
    pub bytes: u64,
-    /// Present only when `include_sstable_rows` was requested. Off by
+    /// Present only when `include_generation_rows` was requested. Off by
    /// default because each count opens an uncached Lance dataset, and the
    /// checkpoint loop polls this route needing only generation numbers.
    #[serde(default)]
@@ -34,11 +34,11 @@ pub struct MemtableStats {
    pub indexes: Vec<String>,
 }

-/// Live state of one table_shard. A table is N table_shards on one node; flattening to
-/// a single number hides the one hot table_shard that is usually why someone
+/// Live state of one bucket. A table is N buckets on one node; flattening to
+/// a single number hides the one hot bucket that is usually why someone
 /// opened this endpoint.
 #[derive(Debug, Clone, Deserialize)]
-pub struct TableShardStats {
+pub struct BucketStats {
    pub shard_id: String,
    /// `Active` | `Sealed` (drop-table 2PC in flight).
    pub status: String,
@@ -47,42 +47,42 @@ pub struct TableShardStats {
    pub current_generation: u64,
    pub replay_after_wal_entry_position: u64,
    pub wal_entry_position_last_seen: u64,
-    pub sstables: Vec<SsTableStats>,
-    /// Whether a pass owns this table_shard's compaction latch right now. Says *a*
+    pub generations: Vec<GenerationStats>,
+    /// Whether a pass owns this bucket's compaction latch right now. Says *a*
    /// driver is running, not *whose*, and the latch is held from dispatch —
    /// including while the pass queues for a pod-wide compactor permit. Read
    /// it as "do not pile on", never as "mine is progressing".
    pub compacting: bool,
-    /// Oldest first, active last. Absent for a `Sealed` table_shard, whose
+    /// Oldest first, active last. Absent for a `Sealed` bucket, whose
    /// in-memory state is torn down.
    #[serde(default)]
    pub memtables: Option<Vec<MemtableStats>>,
 }

-impl TableShardStats {
-    /// The newest SSTable generation, or `None` when the tier is empty.
-    pub(crate) fn newest_sstable_generation(&self) -> Option<u64> {
-        self.sstables.iter().map(|g| g.generation).max()
+impl BucketStats {
+    /// The newest flushed generation, or `None` when L0 is empty.
+    pub(crate) fn newest_generation(&self) -> Option<u64> {
+        self.generations.iter().map(|g| g.generation).max()
    }

-    /// How many SSTables at or below `target` are still uncompacted.
+    /// How many generations at or below `target` are still in L0.
    ///
    /// A count, not a boolean: one pass drains a bounded prefix rather than
    /// the whole target set, so a boolean would read as "no progress" for
    /// every pass but the last. Compaction drains oldest-first, so this
    /// decreases monotonically.
-    pub(crate) fn outstanding_sstables(&self, target: u64) -> usize {
-        self.sstables
+    pub(crate) fn outstanding_generations(&self, target: u64) -> usize {
+        self.generations
            .iter()
            .filter(|g| g.generation <= target)
            .count()
    }
 }

-/// Live LSM state, one entry per table_shard.
+/// Live LSM state, one entry per bucket.
 #[derive(Debug, Clone, Deserialize)]
 pub struct LsmStats {
-    pub table_shards: Vec<TableShardStats>,
+    pub buckets: Vec<BucketStats>,
 }

 /// Server-side JSON envelope for `get_lsm_stats`. `lsm_stats` is null when
@@ -97,18 +97,18 @@ pub(crate) struct GetLsmStatsResponse {
 mod tests {
    use super::*;

-    fn table_shard(shard: &str, sstables: &[u64], compacting: bool) -> TableShardStats {
-        TableShardStats {
+    fn bucket(shard: &str, generations: &[u64], compacting: bool) -> BucketStats {
+        BucketStats {
            shard_id: shard.into(),
            status: "Active".into(),
            writer_epoch: 1,
            manifest_version: 1,
-            current_generation: sstables.iter().max().copied().unwrap_or(0) + 1,
+            current_generation: generations.iter().max().copied().unwrap_or(0) + 1,
            replay_after_wal_entry_position: 0,
            wal_entry_position_last_seen: 0,
-            sstables: sstables
+            generations: generations
                .iter()
-                .map(|g| SsTableStats {
+                .map(|g| GenerationStats {
                    generation: *g,
                    bytes: 1,
                    rows: None,
@@ -123,46 +123,40 @@ mod tests {
    /// generation created after it must not hold the loop open — that is why
    /// the predicate terminates under write load.
    #[test]
-    fn newer_sstables_do_not_extend_the_target() {
-        let start = table_shard("b0", &[7, 8], false);
-        let target = start
-            .newest_sstable_generation()
-            .expect("the SSTable tier is non-empty");
+    fn newer_generations_do_not_extend_the_target() {
+        let start = bucket("b0", &[7, 8], false);
+        let target = start.newest_generation().expect("L0 is non-empty");
        assert_eq!(target, 8);

        // Compaction drained 7 and 8; 9 and 10 arrived while it ran.
-        let later = table_shard("b0", &[9, 10], false);
+        let later = bucket("b0", &[9, 10], false);
        assert_eq!(
-            later.outstanding_sstables(target),
+            later.outstanding_generations(target),
            0,
-            "sstables above the target are somebody else's problem"
+            "generations above the target are somebody else's problem"
        );

        // Still holding 8 means still outstanding.
        assert_eq!(
-            table_shard("b0", &[8, 9], false).outstanding_sstables(target),
+            bucket("b0", &[8, 9], false).outstanding_generations(target),
            1
        );
    }

-    /// The metric counts SSTables, not table shards: a pass drains a bounded
-    /// prefix, so one table_shard going 3 → 2 → 1 → 0 is three steps.
+    /// The metric counts generations, not buckets: a pass drains a bounded
+    /// prefix, so one bucket going 3 → 2 → 1 → 0 is three steps.
    #[test]
-    fn progress_is_measured_in_sstables() {
+    fn progress_is_measured_in_generations() {
        let target = 3;
        let counts: Vec<usize> = [&[1u64, 2, 3][..], &[2, 3][..], &[3][..], &[][..]]
            .iter()
-            .map(|gens| table_shard("b0", gens, false).outstanding_sstables(target))
+            .map(|gens| bucket("b0", gens, false).outstanding_generations(target))
            .collect();
        assert_eq!(counts, vec![3, 2, 1, 0]);
    }

    #[test]
-    fn an_empty_sstable_tier_has_no_target() {
-        assert!(
-            table_shard("b0", &[], false)
-                .newest_sstable_generation()
-                .is_none()
-        );
+    fn empty_l0_has_no_target() {
+        assert!(bucket("b0", &[], false).newest_generation().is_none());
    }
 }
@@ -15,7 +15,7 @@ use lance_index::optimize::OptimizeOptions;
 use log::info;

 pub use chrono::Duration;
-pub use lance::dataset::optimize::CompactionOptions;
+pub use lance::dataset::optimize::{CompactionMode, CompactionOptions, IndexRemapMode};

 use super::NativeTable;
 use crate::error::Result;
@@ -7,6 +7,16 @@
 //! therefore idempotent and does not observe input mutation -- once a row is
 //! filled, changing what the expression reads leaves the stored result alone.
 //!
+//! A column's computed inputs are filled first -- the dependency graph is
+//! walked once, each reachable column filled once in dependency order, each
+//! fill its own commit. Every fill in the pass, the requested column's
+//! included, covers only the fragments of the snapshot the pass started
+//! from: a commit may rebase over a concurrent append, and the fragment that
+//! admits carries placeholder nulls no earlier fill covered, so it waits for
+//! a later refresh rather than being read as values. Two concurrent fills of
+//! one input collide on its field in lance's conflict check, so a dependent
+//! fill can only commit over inputs that were durable when it read them.
+//!
 //! Two passes per fragment. The first scans only the unfilled live rows and
 //! evaluates the expression over them, which yields the exact fill count and
 //! decides whether the fragment is staged at all -- a fragment where nothing
@@ -19,10 +29,14 @@
 //! inputs masked to null first, so a poison value in a row nobody is filling
 //! cannot fail the refresh.

+use std::collections::HashSet;
 use std::sync::Arc;

-use arrow_array::{ArrayRef, BooleanArray, RecordBatch, RecordBatchOptions};
-use arrow_schema::Schema as ArrowSchema;
+use arrow_array::{
+    Array, ArrayRef, BooleanArray, LargeBinaryArray, RecordBatch, RecordBatchOptions, StructArray,
+    new_null_array,
+};
+use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
 use datafusion_expr::ColumnarValue;
 use futures::{Stream, StreamExt, TryStreamExt};
 use lance::Dataset;
@@ -30,7 +44,7 @@ use lance::dataset::WriteDestination;
 use lance::dataset::fragment::FileFragment;
 use lance::dataset::transaction::Operation;
 use lance_core::ROW_ID;
-use lance_core::datatypes::Schema as LanceSchema;
+use lance_core::datatypes::{BlobHandling, Schema as LanceSchema};
 use serde::{Deserialize, Serialize};

 use super::computed_columns::{BoundExpression, ComputedColumnKind, computed_column_from_field};
@@ -41,7 +55,8 @@ use crate::{Error, Result};
 /// The result of refreshing a computed column.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
 pub struct RefreshColumnResult {
-    /// Rows that had a value computed.
+    /// Rows that had a value computed, in the requested column only; inputs
+    /// filled on its behalf are not counted.
    #[serde(default)]
    pub rows_filled: u64,
    /// The commit version associated with the operation.
@@ -52,6 +67,7 @@ pub struct RefreshColumnResult {
 struct RefreshExecution {
    result: RefreshColumnResult,
    source_version: u64,
+    published_version: Option<u64>,
 }

 /// Internal implementation of the refresh logic.
@@ -74,7 +90,12 @@ async fn execute_refresh_column_with_source(

    let expression = declared_expression(&dataset, column)?;
    let schema = Arc::new(ArrowSchema::from(dataset.schema()));
-    let bound = Arc::new(super::computed_columns::bind(schema, column, &expression)?);
+    let bound = Arc::new(super::computed_columns::bind(
+        schema.clone(),
+        column,
+        &expression,
+    )?);
+    ensure_inputs_filled(&dataset, &schema, column, &bound).await?;
    let field = dataset
        .schema()
        .field(column)
@@ -87,6 +108,7 @@ async fn execute_refresh_column_with_source(
        fields: vec![field.clone()],
        metadata: Default::default(),
    };
+    let output_is_blob = field.is_blob_v2();

    let mut rows_filled = 0u64;
    let mut replacements = Vec::new();
@@ -96,29 +118,30 @@ async fn execute_refresh_column_with_source(
            continue;
        }
        rows_filled += gained;
-        let values = fill_stream(&dataset, &fragment, bound.clone(), column).await?;
+        let values =
+            fill_stream(&dataset, &fragment, bound.clone(), column, output_is_blob).await?;
        replacements.push(fragment.write_columns(values, &column_schema).await?);
    }

+    let source_version = dataset.version().version;
    if replacements.is_empty() {
-        let source_version = dataset.version().version;
        return Ok(RefreshExecution {
            result: RefreshColumnResult {
                rows_filled: 0,
                version: source_version,
            },
            source_version,
+            published_version: None,
        });
    }

-    let read_version = dataset.version().version;
    // The dataset's own session, so registrations and caches survive the
    // commit being installed on the handle.
    let session = dataset.session();
    let new_dataset = Dataset::commit(
        WriteDestination::Dataset(dataset.clone()),
        Operation::DataReplacement { replacements },
-        Some(read_version),
+        Some(source_version),
        None,
        None,
        session,
@@ -133,10 +156,52 @@ async fn execute_refresh_column_with_source(
            rows_filled,
            version,
        },
-        source_version: read_version,
+        source_version,
+        published_version: Some(version),
    })
 }

+/// Refuse while a computed input still has rows a refresh of it would fill:
+/// read now, its placeholder null would be evaluated as a value and kept.
+async fn ensure_inputs_filled(
+    dataset: &Dataset,
+    schema: &Arc<ArrowSchema>,
+    column: &str,
+    bound: &BoundExpression,
+) -> Result<()> {
+    for input in &bound.roots {
+        let Some(declaration) = schema
+            .field_with_name(input)
+            .ok()
+            .and_then(computed_column_from_field)
+        else {
+            continue;
+        };
+        let ComputedColumnKind::Sql { expression } = &declaration.kind else {
+            return Err(Error::NotSupported {
+                message: format!(
+                    "computed column '{column}' reads '{input}', whose fill state this \
+                     refresh cannot check; refresh '{input}' first"
+                ),
+            });
+        };
+        let input_bound = super::computed_columns::bind(schema.clone(), input, expression)?;
+        let mut unfilled = 0u64;
+        for fragment in dataset.get_fragments() {
+            unfilled += count_fragment_gains(dataset, &fragment, &input_bound, input).await?;
+        }
+        if unfilled > 0 {
+            return Err(Error::InvalidInput {
+                message: format!(
+                    "computed column '{column}' reads '{input}', which has {unfilled} unfilled \
+                     rows; refresh '{input}' first"
+                ),
+            });
+        }
+    }
+    Ok(())
+}
+
 /// Run the refresh as a [`Job`] in this process.
 pub(crate) async fn execute_refresh_column_async(
    table: &NativeTable,
@@ -160,8 +225,7 @@ pub(crate) async fn execute_refresh_column_async(
            rows_failed: 0,
            rows_remaining: 0,
            source_version: execution.source_version,
-            published_version: (execution.result.rows_filled > 0)
-                .then_some(execution.result.version),
+            published_version: execution.published_version,
        })
    })))
 }
@@ -236,12 +300,15 @@ fn evaluation_batch(
    mask_out: Option<&BooleanArray>,
 ) -> lance_core::Result<RecordBatch> {
    let mut columns = Vec::with_capacity(bound.roots.len());
+    let mut fields = Vec::with_capacity(bound.roots.len());
    for name in &bound.roots {
-        let column = batch.column_by_name(name).ok_or_else(|| {
+        let index = batch.schema_ref().index_of(name).map_err(|_| {
            lance_core::Error::invalid_input(format!(
                "refreshing a computed column read no {name} column"
            ))
        })?;
+        let column = batch.column(index);
+        fields.push(batch.schema_ref().field(index).clone());
        // Rows outside the mask must not reach the expression: a value in a
        // deleted or already-filled row can be one it would choke on.
        columns.push(match mask_out {
@@ -250,7 +317,7 @@ fn evaluation_batch(
        });
    }
    Ok(RecordBatch::try_new_with_options(
-        bound.read_schema.clone(),
+        Arc::new(ArrowSchema::new(fields)),
        columns,
        &RecordBatchOptions::new().with_row_count(Some(batch.num_rows())),
    )?)
@@ -271,6 +338,99 @@ fn evaluate(bound: &BoundExpression, batch: &RecordBatch) -> lance_core::Result<
    }
 }

+fn materialized_blob_ids(schema: &LanceSchema, paths: &[String]) -> Result<HashSet<u32>> {
+    paths
+        .iter()
+        .map(|path| {
+            let field = schema
+                .resolve(path)
+                .and_then(|fields| fields.last().copied())
+                .ok_or_else(|| Error::InvalidInput {
+                    message: format!("computed Blob input '{path}' no longer exists"),
+                })?;
+            if !field.is_blob_v2() {
+                return Err(Error::InvalidInput {
+                    message: format!("computed Blob input '{path}' is no longer Blob v2"),
+                });
+            }
+            u32::try_from(field.id).map_err(|_| Error::InvalidInput {
+                message: format!(
+                    "computed Blob input '{path}' has invalid field id {}",
+                    field.id
+                ),
+            })
+        })
+        .collect()
+}
+
+fn configure_blob_inputs(
+    scanner: &mut lance::dataset::scanner::Scanner,
+    schema: &LanceSchema,
+    bound: &BoundExpression,
+    extra_blob_id: Option<u32>,
+) -> Result<()> {
+    let mut ids = materialized_blob_ids(schema, &bound.blob_paths)?;
+    ids.extend(extra_blob_id);
+    scanner.blob_handling(BlobHandling::SomeBlobsBinary(ids));
+    Ok(())
+}
+
+fn blob_array_from_binary(
+    array: &ArrayRef,
+    target_field: &ArrowField,
+) -> lance_core::Result<ArrayRef> {
+    let values = array
+        .as_any()
+        .downcast_ref::<LargeBinaryArray>()
+        .ok_or_else(|| {
+            lance_core::Error::invalid_input(format!(
+                "a Blob v2 computed output produced {}, expected LargeBinary",
+                array.data_type()
+            ))
+        })?;
+    let mut builder = lance::blob::BlobArrayBuilder::new(values.len());
+    for index in 0..values.len() {
+        if values.is_null(index) {
+            builder.push_null()?;
+        } else {
+            builder.push_bytes(values.value(index))?;
+        }
+    }
+    let minimal = builder.finish()?;
+    let minimal = minimal
+        .as_any()
+        .downcast_ref::<StructArray>()
+        .ok_or_else(|| lance_core::Error::internal("Blob builder returned a non-struct array"))?;
+    let DataType::Struct(target_fields) = target_field.data_type() else {
+        return Err(lance_core::Error::invalid_input(format!(
+            "Blob v2 output field '{}' has non-struct type {}",
+            target_field.name(),
+            target_field.data_type()
+        )));
+    };
+    let columns = target_fields
+        .iter()
+        .map(|field| match field.name().as_str() {
+            "data" | "uri" => minimal
+                .column_by_name(field.name())
+                .cloned()
+                .ok_or_else(|| {
+                    lance_core::Error::internal(format!("Blob builder omitted '{}'", field.name()))
+                }),
+            "position" | "size" => Ok(new_null_array(field.data_type(), minimal.len())),
+            name => Err(lance_core::Error::invalid_input(format!(
+                "Blob v2 output field '{}' has unsupported logical child '{name}'",
+                target_field.name()
+            ))),
+        })
+        .collect::<lance_core::Result<Vec<_>>>()?;
+    Ok(Arc::new(StructArray::try_new(
+        target_fields.clone(),
+        columns,
+        minimal.nulls().cloned(),
+    )?))
+}
+
 /// How many rows of one fragment would gain a value.
 ///
 /// Scans only the unfilled live rows -- deleted rows never reach the
@@ -289,6 +449,7 @@ async fn count_fragment_gains(
        .with_row_id()
        .filter(&format!("{} IS NULL", quote_identifier(column)))?
        .project(&bound.roots)?;
+    configure_blob_inputs(&mut scanner, dataset.schema(), bound, None)?;

    let mut gained = 0u64;
    let mut batches = scanner.try_into_stream().await?;
@@ -310,6 +471,7 @@ async fn fill_stream(
    fragment: &FileFragment,
    bound: Arc<BoundExpression>,
    column: &str,
+    output_is_blob: bool,
 ) -> Result<impl Stream<Item = lance_core::Result<RecordBatch>> + Send + use<>> {
    let mut projection: Vec<String> = bound.roots.clone();
    projection.push(column.to_string());
@@ -319,6 +481,20 @@ async fn fill_stream(
        .with_row_id()
        .include_deleted_rows()
        .project(&projection)?;
+    let output_blob_id = output_is_blob
+        .then(|| {
+            dataset
+                .schema()
+                .field(column)
+                .and_then(|field| u32::try_from(field.id).ok())
+        })
+        .flatten();
+    configure_blob_inputs(
+        &mut scanner,
+        dataset.schema(),
+        bound.as_ref(),
+        output_blob_id,
+    )?;

    let projected = Arc::new(ArrowSchema::new(vec![
        ArrowSchema::from(dataset.schema())
@@ -354,6 +530,11 @@ async fn fill_stream(

        let computed = evaluate(&bound, &evaluation_batch(&batch, &bound, Some(&keep))?)?;
        let merged = arrow_select::zip::zip(&fill, &computed, existing)?;
+        let merged = if output_is_blob {
+            blob_array_from_binary(&merged, projected.field(0))?
+        } else {
+            merged
+        };
        Ok(RecordBatch::try_new(projected.clone(), vec![merged])?)
    }))
 }
@@ -362,8 +543,12 @@ async fn fill_stream(
 mod tests {
    use std::sync::Arc;

-    use arrow_array::{Int32Array, record_batch};
+    use arrow_array::{
+        Array, ArrayRef, Int32Array, LargeBinaryArray, RecordBatch, StructArray, record_batch,
+    };
+    use arrow_schema::Field as ArrowField;
    use futures::TryStreamExt;
+    use lance_core::ROW_ID;

    use crate::connect;
    use crate::query::{ExecutableQuery, QueryBase, Select};
@@ -384,7 +569,8 @@ mod tests {
            .version)
    }

-    async fn read(table: &Table, column: &str) -> Vec<Option<i32>> {
+    async fn read(table: &Table, column: &str) -> Vec<Option<i64>> {
+        use arrow_array::{Array, Int64Array};
        let batches = table
            .query()
            .select(Select::columns(&[column]))
@@ -394,15 +580,19 @@ mod tests {
            .try_collect::<Vec<_>>()
            .await
            .unwrap();
-        let mut values: Vec<Option<i32>> = batches
+        let mut values: Vec<Option<i64>> = batches
            .iter()
            .flat_map(|batch| {
-                batch[column]
-                    .as_any()
-                    .downcast_ref::<Int32Array>()
-                    .unwrap()
-                    .iter()
-                    .collect::<Vec<_>>()
+                let array = &batch[column];
+                match array.as_any().downcast_ref::<Int32Array>() {
+                    Some(ints) => ints.iter().map(|v| v.map(i64::from)).collect::<Vec<_>>(),
+                    None => array
+                        .as_any()
+                        .downcast_ref::<Int64Array>()
+                        .unwrap()
+                        .iter()
+                        .collect::<Vec<_>>(),
+                }
            })
            .collect();
        values.sort();
@@ -414,6 +604,117 @@ mod tests {
        table.add(batch).execute().await.unwrap();
    }

+    #[test]
+    fn test_blob_output_matches_complete_logical_field() {
+        let values: ArrayRef = Arc::new(LargeBinaryArray::from(vec![
+            Some(b"hello".as_slice()),
+            None,
+        ]));
+        let field = ArrowField::new(
+            "image",
+            lance_core::datatypes::BLOB_V2_LOGICAL_TYPE.clone(),
+            true,
+        );
+
+        let output = super::blob_array_from_binary(&values, &field).unwrap();
+        assert_eq!(output.data_type(), field.data_type());
+        let output = output.as_any().downcast_ref::<StructArray>().unwrap();
+        assert_eq!(output.column_by_name("position").unwrap().null_count(), 2);
+        assert_eq!(output.column_by_name("size").unwrap().null_count(), 2);
+    }
+
+    /// The gate's reproducer: `b = coalesce(a, 0)` refreshed before `a`
+    /// must not bake zeros from `a`'s placeholder null. It is refused, and
+    /// names the input, until `a` is filled -- after every append too.
+    #[tokio::test]
+    async fn test_dependent_refresh_refuses_an_unfilled_input() {
+        let table = table_with("dependent_refresh_order", vec![1, 2, 3]).await;
+        table
+            .add_columns()
+            .computed("a", "x + 1")
+            .computed("b", "coalesce(a, 0)")
+            .execute()
+            .await
+            .unwrap();
+
+        let err = table.refresh_column("b").await.unwrap_err();
+        assert!(
+            matches!(&err, Error::InvalidInput { message } if message.contains("refresh 'a' first")),
+            "{err}"
+        );
+        assert_eq!(read(&table, "b").await, vec![None, None, None]);
+
+        assert_eq!(table.refresh_column("a").await.unwrap().rows_filled, 3);
+        assert_eq!(table.refresh_column("b").await.unwrap().rows_filled, 3);
+        assert_eq!(read(&table, "b").await, vec![Some(2), Some(3), Some(4)]);
+
+        append(&table, vec![10]).await;
+        assert!(table.refresh_column("b").await.is_err());
+        table.refresh_column("a").await.unwrap();
+        assert_eq!(table.refresh_column("b").await.unwrap().rows_filled, 1);
+        assert_eq!(
+            table.count_rows(Some("b = 0".to_string())).await.unwrap(),
+            0
+        );
+    }
+
+    /// Names that need quoting, and a nested input, survive the trip through
+    /// declaration metadata and the dependency check: the recorded inputs
+    /// are matched by name, never re-parsed as SQL.
+    #[tokio::test]
+    async fn test_dependent_refresh_handles_awkward_column_names() {
+        use arrow_array::{Int32Array, StructArray};
+        use arrow_schema::{DataType, Field, Fields};
+
+        let conn = connect("memory://").execute().await.unwrap();
+        let age_fields = Fields::from(vec![Field::new("age", DataType::Int32, true)]);
+        let meta = StructArray::new(
+            age_fields.clone(),
+            vec![Arc::new(Int32Array::from(vec![10, 20])) as _],
+            None,
+        );
+        let schema = Arc::new(arrow_schema::Schema::new(vec![
+            Field::new("camelCase", DataType::Int32, true),
+            Field::new("with-hyphen", DataType::Int32, true),
+            Field::new("meta", DataType::Struct(age_fields), true),
+        ]));
+        let batch = arrow_array::RecordBatch::try_new(
+            schema,
+            vec![
+                Arc::new(Int32Array::from(vec![1, 2])) as _,
+                Arc::new(Int32Array::from(vec![100, 200])) as _,
+                Arc::new(meta) as _,
+            ],
+        )
+        .unwrap();
+        let table = conn
+            .create_table("awkward_names", batch)
+            .execute()
+            .await
+            .unwrap();
+
+        table
+            .add_columns()
+            .computed("y", "`camelCase` * 2")
+            .computed("z", "coalesce(y, 0) + `with-hyphen` + meta.age")
+            .execute()
+            .await
+            .unwrap();
+        let z = crate::table::computed_columns::computed_columns(
+            table.schema().await.unwrap().as_ref(),
+        )
+        .into_iter()
+        .find(|c| c.name == "z")
+        .unwrap();
+        assert_eq!(z.inputs, vec!["meta.age", "with-hyphen", "y"]);
+
+        let err = table.refresh_column("z").await.unwrap_err();
+        assert!(err.to_string().contains("refresh 'y' first"), "{err}");
+        assert_eq!(table.refresh_column("y").await.unwrap().rows_filled, 2);
+        assert_eq!(table.refresh_column("z").await.unwrap().rows_filled, 2);
+        assert_eq!(read(&table, "z").await, vec![Some(112), Some(224)]);
+    }
+
    #[tokio::test]
    async fn test_refresh_fills_a_declared_column() {
        let table = table_with("refresh_fills", vec![1, 2, 3]).await;
@@ -651,7 +952,8 @@ mod tests {

        let read_back = read(&table, "doubled").await;
        assert_eq!(read_back.len(), 20_000);
-        let mut expected: Vec<Option<i32>> = values.iter().map(|v| Some(v * 2)).collect();
+        let mut expected: Vec<Option<i64>> =
+            values.iter().map(|v| Some(i64::from(v * 2))).collect();
        expected.sort();
        assert_eq!(read_back, expected);
    }
@@ -1008,4 +1310,366 @@ mod tests {
        let err = table.refresh_column("embedding").await.unwrap_err();
        assert!(matches!(err, Error::NotSupported { message } if message.contains("udf")));
    }
+
+    fn blob_batch(ids: Vec<i32>, payloads: Vec<Option<&[u8]>>) -> RecordBatch {
+        use arrow_array::Int32Array;
+        use arrow_schema::{Field, Schema};
+
+        let mut builder = lance::blob::BlobArrayBuilder::new(payloads.len());
+        for payload in payloads {
+            match payload {
+                Some(payload) => builder.push_bytes(payload).unwrap(),
+                None => builder.push_null().unwrap(),
+            }
+        }
+        RecordBatch::try_new(
+            Arc::new(Schema::new(vec![
+                Field::new("id", arrow_schema::DataType::Int32, false),
+                crate::blob("image", true),
+            ])),
+            vec![Arc::new(Int32Array::from(ids)), builder.finish().unwrap()],
+        )
+        .unwrap()
+    }
+
+    async fn create_blob_table(path: &std::path::Path, batch: RecordBatch) -> Table {
+        let conn = connect(path.to_str().unwrap()).execute().await.unwrap();
+        conn.create_table("blobs", batch).execute().await.unwrap()
+    }
+
+    #[tokio::test]
+    async fn test_refresh_inherits_and_publishes_blob_output() {
+        use arrow_array::UInt64Array;
+        use lance_arrow::{
+            BLOB_DEDICATED_SIZE_THRESHOLD_META_KEY, BLOB_INLINE_SIZE_THRESHOLD_META_KEY,
+        };
+        use lance_core::datatypes::BlobKind;
+
+        use crate::table::schema_evolution::FieldMetadataUpdate;
+
+        let tmp = tempfile::tempdir().unwrap();
+        let table = create_blob_table(
+            tmp.path(),
+            blob_batch(
+                vec![1, 2, 3, 4],
+                vec![Some(b"hello"), Some(b"ab"), Some(b""), None],
+            ),
+        )
+        .await;
+        table
+            .add_columns()
+            .computed("image_copy", "image")
+            .execute()
+            .await
+            .unwrap();
+        table
+            .update_field_metadata(&[FieldMetadataUpdate::new("image_copy")
+                .set(BLOB_INLINE_SIZE_THRESHOLD_META_KEY, "1")
+                .set(BLOB_DEDICATED_SIZE_THRESHOLD_META_KEY, "4")])
+            .await
+            .unwrap();
+
+        let first_refresh = table.refresh_column("image_copy").await.unwrap();
+        assert_eq!(first_refresh.rows_filled, 3);
+        assert_eq!(
+            table.blob_columns().await.unwrap(),
+            vec!["image".to_string(), "image_copy".to_string()]
+        );
+
+        let batches = table
+            .query()
+            .with_row_id()
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let batch = arrow_select::concat::concat_batches(&batches[0].schema(), &batches).unwrap();
+        assert!(
+            batch
+                .column_by_name("image_copy")
+                .unwrap()
+                .as_any()
+                .is::<arrow_array::StructArray>()
+        );
+        let row_ids = batch
+            .column_by_name(ROW_ID)
+            .unwrap()
+            .as_any()
+            .downcast_ref::<UInt64Array>()
+            .unwrap()
+            .values()
+            .to_vec();
+        let original = table.fetch_blobs("image", &row_ids).await.unwrap();
+        let copied = table.fetch_blobs("image_copy", &row_ids).await.unwrap();
+        assert_eq!(original, copied);
+        let ids = batch
+            .column_by_name("id")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .unwrap();
+        let files = table
+            .fetch_blob_files("image_copy", &row_ids)
+            .await
+            .unwrap();
+        let mut layouts = ids
+            .values()
+            .iter()
+            .copied()
+            .zip(files)
+            .map(|(id, file)| (id, file.and_then(|file| file.kind())))
+            .collect::<Vec<_>>();
+        layouts.sort_by_key(|(id, _)| *id);
+        assert_eq!(
+            layouts,
+            vec![
+                (1, Some(BlobKind::Dedicated)),
+                (2, Some(BlobKind::Packed)),
+                (3, Some(BlobKind::Inline)),
+                (4, None),
+            ]
+        );
+
+        table
+            .add(blob_batch(vec![5], vec![Some(b"appended")]))
+            .execute()
+            .await
+            .unwrap();
+        table
+            .optimize(crate::table::OptimizeAction::Compact {
+                options: crate::table::CompactionOptions::default(),
+                remap_options: None,
+            })
+            .await
+            .unwrap();
+        assert_eq!(
+            table
+                .refresh_column("image_copy")
+                .await
+                .unwrap()
+                .rows_filled,
+            1
+        );
+        assert_eq!(
+            table
+                .refresh_column("image_copy")
+                .await
+                .unwrap()
+                .rows_filled,
+            0
+        );
+
+        table.checkout(first_refresh.version).await.unwrap();
+        assert_eq!(table.count_rows(None).await.unwrap(), 4);
+        assert_eq!(
+            table.blob_columns().await.unwrap(),
+            vec!["image".to_string(), "image_copy".to_string()]
+        );
+        table.checkout_latest().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_refresh_inherits_nested_struct_blob_input() {
+        use arrow_array::{Int32Array, StructArray, UInt64Array};
+        use arrow_schema::{DataType, Field, Fields, Schema};
+
+        let tmp = tempfile::tempdir().unwrap();
+        let mut blob_builder = lance::blob::BlobArrayBuilder::new(2);
+        blob_builder.push_bytes(b"nested").unwrap();
+        blob_builder.push_null().unwrap();
+        let blob_field = crate::blob("image", true);
+        let metadata_fields = Fields::from(vec![blob_field.clone()]);
+        let metadata = StructArray::new(
+            metadata_fields.clone(),
+            vec![blob_builder.finish().unwrap()],
+            None,
+        );
+        let batch = RecordBatch::try_new(
+            Arc::new(Schema::new(vec![
+                Field::new("id", DataType::Int32, false),
+                Field::new("metadata", DataType::Struct(metadata_fields), true),
+            ])),
+            vec![Arc::new(Int32Array::from(vec![1, 2])), Arc::new(metadata)],
+        )
+        .unwrap();
+        let table = create_blob_table(tmp.path(), batch).await;
+        table
+            .add_columns()
+            .computed("payload_copy", "metadata.image")
+            .execute()
+            .await
+            .unwrap();
+
+        assert_eq!(
+            table
+                .refresh_column("payload_copy")
+                .await
+                .unwrap()
+                .rows_filled,
+            1
+        );
+        assert_eq!(
+            table.blob_columns().await.unwrap(),
+            vec!["metadata.image".to_string(), "payload_copy".to_string()]
+        );
+        let batches = table
+            .query()
+            .with_row_id()
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let row_ids = batches[0]
+            .column_by_name(ROW_ID)
+            .unwrap()
+            .as_any()
+            .downcast_ref::<UInt64Array>()
+            .unwrap()
+            .values();
+        let payloads = table.fetch_blobs("payload_copy", row_ids).await.unwrap();
+        assert_eq!(payloads.value(0), b"nested");
+        assert!(payloads.is_null(1));
+    }
+
+    #[tokio::test]
+    async fn test_refresh_preserves_list_shape_when_materializing_blob_input() {
+        use arrow_array::{Int32Array, ListArray};
+        use arrow_buffer::{OffsetBuffer, ScalarBuffer};
+        use arrow_schema::{DataType, Field, Schema};
+
+        let tmp = tempfile::tempdir().unwrap();
+        let mut blob_builder = lance::blob::BlobArrayBuilder::new(3);
+        blob_builder.push_bytes(b"a").unwrap();
+        blob_builder.push_bytes(b"bb").unwrap();
+        blob_builder.push_null().unwrap();
+        let item = Arc::new(crate::blob("item", true));
+        let images = ListArray::new(
+            item.clone(),
+            OffsetBuffer::new(ScalarBuffer::from(vec![0, 2, 3])),
+            blob_builder.finish().unwrap(),
+            None,
+        );
+        let batch = RecordBatch::try_new(
+            Arc::new(Schema::new(vec![
+                Field::new("id", DataType::Int32, false),
+                Field::new("images", DataType::List(item), true),
+            ])),
+            vec![Arc::new(Int32Array::from(vec![1, 2])), Arc::new(images)],
+        )
+        .unwrap();
+        let table = create_blob_table(tmp.path(), batch).await;
+        table
+            .add_columns()
+            .computed("image_payloads", "images")
+            .execute()
+            .await
+            .unwrap();
+
+        assert_eq!(
+            table
+                .refresh_column("image_payloads")
+                .await
+                .unwrap()
+                .rows_filled,
+            2
+        );
+        let batches = table
+            .query()
+            .select(Select::columns(&["image_payloads"]))
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let output = batches[0]
+            .column_by_name("image_payloads")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<ListArray>()
+            .unwrap();
+        assert_eq!(output.value_offsets(), &[0, 2, 3]);
+        assert!(output.values().as_any().is::<LargeBinaryArray>());
+    }
+
+    #[tokio::test]
+    async fn test_refresh_inherits_external_blob_input() {
+        use arrow_array::{Int32Array, StringArray, UInt64Array};
+        use arrow_schema::{DataType, Field, Schema};
+
+        let tmp = tempfile::tempdir().unwrap();
+        let payload = b"external-payload";
+        let path = tmp.path().join("payload.bin");
+        std::fs::write(&path, payload).unwrap();
+        let uri = url::Url::from_file_path(path).unwrap().to_string();
+        let conn = connect(tmp.path().join("db").to_str().unwrap())
+            .execute()
+            .await
+            .unwrap();
+        let table = conn
+            .create_empty_table(
+                "external",
+                Arc::new(Schema::new(vec![
+                    Field::new("id", DataType::Int32, false),
+                    crate::blob("image", true),
+                ])),
+            )
+            .execute()
+            .await
+            .unwrap();
+        let batch = RecordBatch::try_new(
+            Arc::new(Schema::new(vec![
+                Field::new("id", DataType::Int32, false),
+                Field::new("image", DataType::Utf8, true),
+            ])),
+            vec![
+                Arc::new(Int32Array::from(vec![1])),
+                Arc::new(StringArray::from(vec![Some(uri)])),
+            ],
+        )
+        .unwrap();
+        table
+            .add(batch)
+            .allow_external_blob_outside_bases(true)
+            .execute()
+            .await
+            .unwrap();
+        table
+            .add_columns()
+            .computed("payload_copy", "image")
+            .execute()
+            .await
+            .unwrap();
+
+        assert_eq!(
+            table
+                .refresh_column("payload_copy")
+                .await
+                .unwrap()
+                .rows_filled,
+            1
+        );
+        let batches = table
+            .query()
+            .with_row_id()
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let row_ids = batches[0]
+            .column_by_name(ROW_ID)
+            .unwrap()
+            .as_any()
+            .downcast_ref::<UInt64Array>()
+            .unwrap()
+            .values();
+        let payloads = table.fetch_blobs("payload_copy", row_ids).await.unwrap();
+        assert_eq!(payloads.value(0), payload);
+    }
 }
Author	SHA1	Message	Date
Gatefixer	9bfd475702	Merge remote-tracking branch 'origin/main' into gatekeeper/fix-2325-1 # Conflicts: # python/python/lancedb/remote/table.py	2026-08-28 09:45:02 +00:00
Xuanwo	0559108fa9	feat: support blob computed column refresh (#4081 ) Computed-column planning currently sees Blob v2 storage descriptors, so expressions cannot consume payload bytes or preserve Blob semantics in their outputs. A computed declaration now derives its output field from its expression. A direct projection of a Blob v2 field inherits the source field's Blob metadata; other expressions retain their ordinary Arrow-inferred type. Declarations remain ordered, so the same rule applies across chained projections. Refresh materializes referenced Blob inputs as `LargeBinary` payload bytes and publishes inherited Blob outputs through Lance's Blob conversion path. Remote requests remain within the shared namespace contract as `{name, computed}`; the server planner is being updated in tandem to implement the same Blob-aware planning semantics, and remote enablement must be aligned with that server rollout. The existing null-as-unfilled contract remains unchanged. Row-level freshness and cell flags remain follow-up work.	2026-08-28 17:23:49 +08:00
Will Jones	6ab3b9eb30	ci: upgrade chacha20 to 0.10.2 (#4078 ) The pinned version was yanked due to UB in some SIMD kernels. Upgrading.	2026-08-27 19:06:07 -07:00
Lance Release	c94d9a2a16	Bump version: 0.38.0-beta.11 → 0.38.0-beta.12	2026-08-28 00:52:01 +00:00
Wyatt Alt	6c8aa22704	feat: let a computed-column batch read its own earlier declarations (#4072 ) `add_columns().computed()` accepted several columns in one call but bound each against the table's schema as it stood before the call, so `a` and `b = a + 1` had to be two commits. A server staging declarations behind other schema work has no atomic way to do that, and a caller reading the builder's plural signature reasonably expects the batch to be one. Each accepted column now joins the schema the next one resolves against, so the batch is planned and committed as one. Order is the dependency order; reading ahead is still an unknown column. `validate_declarations` exposes the schema-level checks -- the Function-binding guard and the planning -- without a commit, for callers that must reject before earlier work in the same request lands; LSM state is table state and stays a commit-time check. Refresh order matters for a dependent column: `b = coalesce(a, 0)` refreshed before `a` would bake zeros from `a`'s placeholder null, and the fill-once contract keeps them. Refresh now refuses, naming the input, while a computed input still has rows a refresh of it would fill -- the same probe refresh already uses to detect a no-op. Otherwise it is one snapshot and one commit, as before; a concurrent append is not in the commit and waits for the next refresh. Refreshing dependencies on the caller's behalf was considered and rejected: it is not how materialized views or our own backfill scheduler behave, and it needs multi-commit fencing that an explicit per-row fill marker would make unnecessary.	2026-08-27 17:47:27 -07:00
Will Jones	84f46df876	ci(nodejs): fix nightly OOM on the aarch64 publish legs (#4077 ) The nightly `NPM Publish` run has failed every night since at least Aug 23, always on the same two legs: `aarch64-unknown-linux-gnu` and `aarch64-unknown-linux-musl`. The other five targets pass. rustc is OOM-killed during the fat-LTO codegen of the cdylib — `signal: 9` with no diagnostic, about 27 minutes in — and on the musl leg that takes the whole runner down with `The runner has received a shutdown signal`. Both legs now pass: \| leg \| before \| peak memory \| wall time \| \| --- \| --- \| --- \| --- \| \| `aarch64-unknown-linux-gnu` \| OOM-killed at ~27 min \| 31391 → 22851 MiB \| 38m43s → 22m04s \| \| `aarch64-unknown-linux-musl` \| runner killed at ~28 min \| >32 GiB → 16516 MiB \| ~40 min → 20m50s \| ThinLTO is most of that. Fat LTO is single-threaded, and its peak is consumed inside rustc's LLVM before any linker process is spawned — which is why it is the whole fix on musl, and why lld alone left the gnu leg still peaking at 31391 MiB against the runner's 32 GiB. Both legs now use the `lto: thin` / `codegen_units: 16` settings that darwin and both Windows legs already use, at a cost of a few percent runtime performance. lld covers the rest, on the gnu leg. arm64 Linux otherwise links through GNU `ld` where x86_64 already defaults to `rust-lld`, which is why only the arm64 legs hit this at all; on a comparable arm64 build (`lancedb/sophon#7313`) it cut the largest single linker process from 7.0 to 4.0 GiB and wall time by 35%. The flags live in a small wrapper script used as the linker rather than in `-C link-arg`, because the per-target rustflags variable does not reach every unit that links: dependency crates linking a dylib (`crc-fast`, `lance-arrow`) were invoked as bare `clang`, which targets the x86_64 host and fails with `Relocations in generic ELF (EM: 183)`. Separately, and affecting five legs rather than two: the three ThinLTO targets exported `CARGO_PROFILE_RELEASE_LTO` and `CARGO_PROFILE_RELEASE_CODEGEN_UNITS` from `pre_build`, which runs inside the build step — after the cache step. `Swatinem/rust-cache` computes its key when the action runs, before any step, so step-local values are invisible to it. The result is a loop that never converges: the key never changes, so restores are exact hits, an exact hit makes the post-run save a no-op, and cargo invalidates the restored artifacts anyway because the flags differ. Those legs have been rebuilding cold on every run. Both values move to job-level `env:` ahead of the cache step, driven by new `lto:`/`codegen_units:` matrix fields, and are forwarded into the containers with `-e` since `docker run` inherits nothing. Every leg's cache key shifts once as a result, so expect one cold rebuild. A `Report peak memory` step is added so whether these legs fit is a number rather than an inference from whether the runner survived. It produced the figures above. ## Not included Moving these legs to native arm64 runners. It would retire the zig cross path, the `AT_HWCAP2` workaround and the `TARGET_CC` override, and arm64 runners are billed roughly 37% below x64 at equal core count — but the `lts-debian-aarch64` image exists to link against the manylinux2014 sysroot's glibc 2.17, and building natively on ubuntu-24.04 would raise the minimum glibc for every published aarch64 binary. That is a user-facing decision, not a CI cleanup. Dropping these legs to smaller runners, which is where the real cost saving is — larger runners are billed even on public repos. On these numbers it is not available yet: musl at 16516 MiB is about 130 MiB over what a 16 GB standard runner has. Worth revisiting as a follow-up. ## Testing Cargo's rustflags precedence was checked locally rather than taken from the docs, since getting it wrong would silently change the published binaries. With a throwaway crate carrying both a `target.'cfg(all())'` and a per-target rustflags table: setting `RUSTFLAGS` discards both, and setting it to the empty string discards them too. That rules out routing the linker flag through a job-level `RUSTFLAGS`, because `env:` keys cannot be conditionally omitted and every other leg would then silently lose the `target-cpu`/`target-feature` settings in `.cargo/config.toml` — `+avx2` on x86_64 and `-crt-static` on aarch64-musl. --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>	2026-08-27 17:35:06 -07:00
Drew	83cff3ab93	fix(python): use one blobv2 type and coerce blob writes by metadata (#4065 )	2026-08-27 17:07:29 -07:00
Will Jones	b85776c22a	fix(listing)!: page table listings from the store's own cursor (#3979 ) BREAKING CHANGE: list_tables now provides tables in arbitrary order and the page token is now completely opaque. `table_names` retains the old behavior of lexical ordering and `start-after` semantics. Listing the tables in a directory database cost what the database held rather than what the page held. `ListingDatabase::list_tables` enumerated every child directory of the base path, sorted the names, then discarded all but the requested page — on every request, for every page. On object storage that is one full listing per page. This PR pages the store instead. `list_tables` asks for one page at a time through `ObjectStore::read_dir_page`, carrying the store's own continuation token, so a page is one request. Non-table children can leave a page short of its limit, so the walk continues until the page is full or the store runs out. --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>	2026-08-27 15:52:41 -07:00
Xuanwo	670bda8725	Merge branch 'main' into gatekeeper/fix-2325-1	2026-08-27 18:13:07 +08:00
Gatefixer	10151b2dc8	Merge remote-tracking branch 'origin/main' into gatekeeper/fix-2325-1	2026-08-22 07:07:55 +00:00
Gatefixer	53b4c3b715	Merge origin/main into gatekeeper/fix-2325-1 # Conflicts: # python/python/lancedb/__init__.py	2026-08-22 07:07:52 +00:00
Gatefixer	638430fdb4	Merge origin/main into gatekeeper/fix-2325-1	2026-08-21 09:23:38 +00:00
Gatefixer	2221b8df6a	Merge origin/main into gatekeeper/fix-2325-1	2026-08-21 08:56:03 +00:00
Gatefixer	14eeae4bd4	Merge origin/main into gatekeeper/fix-2325-1	2026-08-21 08:47:29 +00:00
Gatefixer	320755ed55	Merge origin/main into gatekeeper/fix-2325-1 # Conflicts: # python/python/lancedb/__init__.py	2026-08-21 08:13:23 +00:00
Gatefixer	e55c2da7b1	fix(python): clarify compaction limits	2026-08-20 11:45:51 +00:00
Gatefixer	d33b05328c	Merge remote-tracking branch 'origin/main' into gatekeeper/fix-2325-1	2026-08-20 11:09:16 +00:00
Gatefixer	82f5355b71	fix(python): expose compaction source limits	2026-08-14 22:41:06 +00:00
Gatefixer	40cff9b644	Merge origin/main into gatekeeper/fix-2325-1	2026-08-14 21:59:28 +00:00
Gatefixer	edf95e53fc	Merge origin/main into gatekeeper/fix-2325-1	2026-08-07 19:46:40 +00:00
Gatefixer	0b5eba085d	Merge origin/main into gatekeeper/fix-2325-1	2026-08-07 19:32:06 +00:00
Gatefixer	21bf859c0b	fix(python): validate compaction option bounds	2026-08-06 06:30:38 +00:00
Gatefixer	e0499de959	fix(python): expose optimize compaction options	2026-08-06 05:01:42 +00:00