refactor(lsm): rename LSM stats to SSTable and table shard

Aligns the LSM stats surface with the MemWAL naming settled upstream in lance-format/lance#7943 and #7957, where the persisted unit became an SSTable. Two terms in this API predate that pass. **Generation -> SSTable.** A `GenerationStats` describes one flushed MemTable, which is an SSTable. The generation *number* is kept — an SSTable is identified by its generation — so only the noun moved. **Bucket -> table shard.** Each entry is one MemWAL shard. "Bucket" names only the hash sharding transform, so it was wrong for `identity` and `year` sharding, which produce shards and no buckets at all. | Before | After | |---|---| | `GenerationStats` | `SsTableStats` | | `BucketStats` | `TableShardStats` | | `LsmStats.buckets` | `LsmStats.table_shards` | | `BucketStats.generations` | `TableShardStats.sstables` | | `include_generation_rows` | `include_sstable_rows` | | `newest_generation` | `newest_sstable_generation` | | `outstanding_generations` | `outstanding_sstables` | Applied across Rust, Python, TypeScript, and Java, including the `get_lsm_stats` JSON field names. `LsmWriteSpec::Bucket` and `num_buckets` are unchanged — those name the sharding transform, not the shard. ## Compatibility Breaking for the MemWAL LSM stats API, which is experimental and paired with a server that renames the same fields. The JSON keys `table_shards` and `sstables` must roll out together with the WAL server change. ## Validation - `cargo check -p lancedb --all-features`, `cargo fmt --all` - `ruff format --check` and `ruff check` on the touched Python - `biome check` on the touched TypeScript The typedoc markdown under `docs/src/js` was updated by hand, not regenerated: `npm run docs` needs the napi-built `./native` types. Worth running `npm run docs` on this branch to confirm the generator agrees. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HgVf5C2yRbwK6pe1aMYkmg
fix(node): accept Arrow metadata across JavaScript realms (#3904 )
2026-08-28 08:58:41 +00:00 · 2026-08-27 15:18:13 -05:00 · 2026-08-27 23:35:02 +08:00 · 2026-08-27 20:28:21 +08:00
37 changed files with 1051 additions and 418 deletions
@@ -221,7 +221,7 @@ abstract checkpointLsm(): Promise<void>

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

-Seals once, then triggers compaction and polls until the L0 that existed
+Freezes once, then triggers compaction and polls until the SSTables 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 L0 → base compaction pass per bucket.
+Trigger a background SSTable compaction pass per table shard.

 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>
 ```

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

 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(includeGenerationRows?): Promise<undefined | LsmStats>
+abstract getLsmStats(includeSstableRows?): Promise<undefined | LsmStats>
 ```

-Read live per-bucket LSM state.
+Read live per-table-shard 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

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

 #### Returns
@@ -60,7 +60,6 @@
 - [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)
@@ -87,7 +86,6 @@
 - [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)
@@ -126,7 +124,9 @@
 - [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)
@@ -1,40 +0,0 @@
-[**@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-bucket LSM state, as returned by `Table#getLsmStats`.
+Live per-table-shard LSM state, as returned by `Table#getLsmStats`.

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

 ## Properties

-### buckets
+### tableShards

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

-One entry per bucket backing this table.
+One entry per table shard backing this table.
@@ -0,0 +1,40 @@
+[**@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.
@@ -2,12 +2,12 @@

 ***

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

-# Interface: BucketStats
+# Interface: TableShardStats

-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
+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
 this endpoint.

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

-Whether a pass owns this bucket's compaction latch right now. Says *a*
+Whether a pass owns this table shard'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.

 ***

-### generations
+### sstables

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

-Flushed L0 generations not yet merged into the base table.
+SSTables 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"` bucket, whose
+Oldest first, active last. Absent for a `"Sealed"` table shard, whose
 in-memory state is torn down.

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

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

 ***

@@ -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,
- * seal into L0 generations, and are merged into the base table by compaction.
+ * freeze into SSTables, 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.bucket("id", 16));
+ * lsm.setLsmWriteSpec(LsmWriteSpec.tableShard("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; bucket sharding
+   * <p>All variants require the table to have an unenforced primary key; tableShard sharding
   * additionally requires it to be the single column being bucketed.
   */
  public void setLsmWriteSpec(LsmWriteSpec spec) {
@@ -130,7 +130,7 @@ public class LanceDbTableLsm {
  }

  /**
-   * Seal every bucket's active memtable into a new L0 generation.
+   * Freeze every table shard's active memtable into a new SSTable.
   *
   * <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 L0 → base compaction pass per bucket.
+   * Trigger a background SSTable compaction pass per table shard.
   *
   * <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-bucket LSM state.
+   * Read live per-tableShard LSM state.
   *
-   * <p>Answers "how far behind is my fresh tier", "which bucket is hot", and "why is my fresh-tier
+   * <p>Answers "how far behind is my fresh tier", "which tableShard 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 includeGenerationRows Also count rows per L0 generation. Off by default because each
+   * @param includeSstableRows Also count rows per SSTable. 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 includeGenerationRows) {
+  public Optional<LsmStats> getLsmStats(boolean includeSstableRows) {
    Map<String, Object> body = new LinkedHashMap<String, Object>();
-    body.put("include_generation_rows", includeGenerationRows);
+    body.put("include_sstable_rows", includeSstableRows);
    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>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
+   * <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
   * <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 generations.
+      // no-op, so a re-issue does not churn empty sstables.
      if (issueVoid(this::flushLsm)) {
        backoff(reissue);
        continue;
@@ -220,7 +220,7 @@ public class LanceDbTableLsm {
        return;
      }

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

  /**
-   * Trigger and poll until no bucket holds a generation at or below its target.
+   * Trigger and poll until no tableShard 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 bucket's compaction latch, held from dispatch until the pass
+      // `compacting` is the tableShard'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 (BucketStats bucket : stats.value.get().buckets()) {
-        Long target = targets.get(bucket.shardId());
+      for (TableShardStats tableShard : stats.value.get().tableShards()) {
+        Long target = targets.get(tableShard.shardId());
        if (target == null) {
          continue;
        }
-        long remaining = bucket.outstandingGenerations(target);
+        long remaining = tableShard.outstandingSstables(target);
        if (remaining > 0) {
          outstanding += remaining;
-          allCompacting &= bucket.compacting();
+          allCompacting &= tableShard.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 bucket at all, which the poll
+          // A 429 here means the server could latch no tableShard 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 bucket, skipping buckets holding none. */
-  private static Map<String, Long> newestGenerations(LsmStats stats) {
+  /** The newest generation held by each tableShard, skipping tableShards holding none. */
+  private static Map<String, Long> newestSstableGenerations(LsmStats stats) {
    Map<String, Long> targets = new HashMap<String, Long>();
-    for (BucketStats bucket : stats.buckets()) {
-      OptionalLong newest = bucket.newestGeneration();
+    for (TableShardStats tableShard : stats.tableShards()) {
+      OptionalLong newest = tableShard.newestSstableGeneration();
      if (newest.isPresent()) {
-        targets.put(bucket.shardId(), newest.getAsLong());
+        targets.put(tableShard.shardId(), newest.getAsLong());
      }
    }
    return targets;
@@ -20,37 +20,37 @@ import java.util.Collections;
 import java.util.List;

 /**
- * Live per-bucket LSM state, as returned by {@link LanceDbTableLsm#getLsmStats()}.
+ * Live per-tableShard LSM state, as returned by {@link LanceDbTableLsm#getLsmStats()}.
 *
- * <p>Nothing here is derived: sums and differences (total L0 bytes, WAL lag) are the caller's to
+ * <p>Nothing here is derived: sums and differences (total SSTable 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<BucketStats> buckets;
+  private final List<TableShardStats> tableShards;

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

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

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

  @Override
  public String toString() {
-    return "LsmStats{buckets=" + buckets + "}";
+    return "LsmStats{tableShards=" + tableShards + "}";
  }
 }
@@ -17,21 +17,21 @@ import com.fasterxml.jackson.databind.JsonNode;

 import java.util.OptionalLong;

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

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

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

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

  /**
-   * Rows in this generation, present only when {@code includeGenerationRows} was requested. Off by
+   * Rows in this generation, present only when {@code includeSstableRows} 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 GenerationStats fromJson(JsonNode node) {
+  static SsTableStats fromJson(JsonNode node) {
    JsonFields.requiredObject(node, CONTEXT);
-    return new GenerationStats(
+    return new SsTableStats(
        JsonFields.requiredLong(node, "generation", CONTEXT),
        JsonFields.requiredLong(node, "bytes", CONTEXT),
        JsonFields.optionalLong(node, "rows", CONTEXT));
@@ -59,6 +59,6 @@ public class GenerationStats {

  @Override
  public String toString() {
-    return "GenerationStats{generation=" + generation + ", bytes=" + bytes + ", rows=" + rows + "}";
+    return "SsTableStats{generation=" + generation + ", bytes=" + bytes + ", rows=" + rows + "}";
  }
 }
@@ -22,11 +22,11 @@ import java.util.Optional;
 import java.util.OptionalLong;

 /**
- * 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.
+ * 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.
 */
-public class BucketStats {
-  private static final String CONTEXT = "bucket stats";
+public class TableShardStats {
+  private static final String CONTEXT = "tableShard stats";

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

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

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

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

  /**
-   * Whether a pass owns this bucket's compaction latch right now. Says <em>a</em> driver is
+   * Whether a pass owns this tableShard'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 BucketStats {
    return compacting;
  }

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

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

  /**
-   * How many generations at or below {@code target} are still in L0.
+   * How many SSTables at or below {@code target} are still uncompacted.
   *
   * <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 outstandingGenerations(long target) {
+  long outstandingSstables(long target) {
    long count = 0;
-    for (GenerationStats generation : generations) {
+    for (SsTableStats generation : sstables) {
      if (generation.generation() <= target) {
        count++;
      }
@@ -148,11 +148,11 @@ public class BucketStats {
    return count;
  }

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

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

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

  @Override
  public String toString() {
-    return "BucketStats{shardId="
+    return "TableShardStats{shardId="
        + shardId
        + ", status="
        + status
        + ", currentGeneration="
        + currentGeneration
-        + ", generations="
-        + generations
+        + ", sstables="
+        + sstables
        + ", compacting="
        + compacting
        + "}";
@@ -132,10 +132,10 @@ public class LanceDbTableLsmTest {
    enqueue("set_lsm_write_spec", 200, "");

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

    JsonNode body = MAPPER.readTree(requestBodies.get(0));
-    assertEquals("bucket", body.get("sharding").get("mode").asText());
+    assertEquals("tableShard", 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\":\"bucket\",\"column\":\"id\","
+        "{\"lsm_write_spec\":{\"sharding\":{\"mode\":\"tableShard\",\"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(bucket("shard-0", false, 7L, 8L)));
+    enqueue("get_lsm_stats", 200, stats(tableShard("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_generation_rows").asBoolean());
+    assertTrue(MAPPER.readTree(requestBodies.get(0)).get("include_sstable_rows").asBoolean());
    assertTrue(got.isPresent());
-    BucketStats decoded = got.get().buckets().get(0);
+    TableShardStats decoded = got.get().tableShards().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.generations().get(0).bytes());
-    assertFalse(decoded.generations().get(0).rows().isPresent(), "rows absent unless requested");
+    assertEquals(1024, decoded.sstables().get(0).bytes());
+    assertFalse(decoded.sstables().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\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+        "{\"lsm_stats\":{\"tableShards\":[{\"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,"
-            + "\"generations\":[{\"generation\":7,\"bytes\":1024,\"rows\":42}],"
+            + "\"sstables\":[{\"generation\":7,\"bytes\":1024,\"rows\":42}],"
            + "\"compacting\":true,\"memtables\":[{\"generation\":8,\"rows\":5,"
            + "\"bytes\":64,\"batches\":2,\"indexes\":[\"id_idx\"]}]}]}}");

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

    assertEquals(3, decoded.replayAfterWalEntryPosition());
    assertEquals(11, decoded.walEntryPositionLastSeen());
    assertTrue(decoded.compacting());
-    assertEquals(42, decoded.generations().get(0).rows().getAsLong());
+    assertEquals(42, decoded.sstables().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_generation_rows").asBoolean());
+    assertFalse(MAPPER.readTree(requestBodies.get(0)).get("include_sstable_rows").asBoolean());
  }

  // ===========================================================================
@@ -334,8 +334,8 @@ public class LanceDbTableLsmTest {
  @Test
  public void testCheckpointReturnsWhenNoGenerationsOutstanding() {
    enqueue("flush_lsm", 200, "");
-    // A bucket with no L0 generations yields no target, so the drain never starts.
-    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false)));
+    // A table shard with no SSTables yields no target, so the drain never starts.
+    enqueue("get_lsm_stats", 200, stats(tableShard("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 generations 7 and 8, so target = 8.
-    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
+    // Watermark read: shard-0 holds sstables 7 and 8, so target = 8.
+    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
    // First drain poll: both still outstanding, nothing compacting -> dispatch a pass.
-    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
+    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
    // Second drain poll: drained past the target -> done.
-    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 9L)));
+    enqueue("get_lsm_stats", 200, stats(tableShard("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(bucket("shard-0", true, 4L)));
+    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", true, 4L)));
    // Still compacting on the first poll, so no pass is dispatched; then it drains.
-    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", true, 4L)));
-    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 5L)));
+    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", true, 4L)));
+    enqueue("get_lsm_stats", 200, stats(tableShard("shard-0", false, 5L)));

    lsm.checkpointLsm();

-    assertEquals(0, countCalls("compact_lsm"), "a latched bucket is left alone");
+    assertEquals(0, countCalls("compact_lsm"), "a latched tableShard 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(bucket("shard-0", false)));
+    enqueue("get_lsm_stats", 200, stats(tableShard("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(bucket("shard-0", false)));
+    enqueue("get_lsm_stats", 200, stats(tableShard("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 buckets", which is indistinguishable from a drained table, so {@code checkpointLsm}
+   * read as "no tableShards", 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 buckets", "{\"lsm_stats\":{}}");
-    malformed.put("bucket missing its required fields", "{\"lsm_stats\":{\"buckets\":[{}]}}");
+    malformed.put("stats object with no tableShards", "{\"lsm_stats\":{}}");
+    malformed.put("tableShard missing its required fields", "{\"lsm_stats\":{\"tableShards\":[{}]}}");
    malformed.put(
-        "bucket missing generations",
-        "{\"lsm_stats\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+        "tableShard missing sstables",
+        "{\"lsm_stats\":{\"tableShards\":[{\"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\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+        "{\"lsm_stats\":{\"tableShards\":[{\"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,"
-            + "\"generations\":[{\"generation\":\"7\",\"bytes\":1024}],"
+            + "\"sstables\":[{\"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(BucketStats bucket) {
+  private static List<Long> generationNumbers(TableShardStats tableShard) {
    List<Long> numbers = new ArrayList<Long>();
-    for (GenerationStats generation : bucket.generations()) {
+    for (SsTableStats generation : tableShard.sstables()) {
      numbers.add(generation.generation());
    }
    return numbers;
  }

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

-  private static String bucket(String shardId, boolean compacting, Long... generations) {
+  private static String tableShard(String shardId, boolean compacting, Long... sstables) {
    StringBuilder gens = new StringBuilder();
-    for (Long generation : generations) {
+    for (Long generation : sstables) {
      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,\"generations\":["
+        + "\"wal_entry_position_last_seen\":0,\"sstables\":["
        + gens
        + "],\"compacting\":"
        + compacting
@@ -1,5 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
+import * as fs from "node:fs";
+import * as vm from "node:vm";
 import * as arrow15 from "apache-arrow-15";
 import * as arrow16 from "apache-arrow-16";
 import * as arrow17 from "apache-arrow-17";
@@ -40,6 +42,41 @@ function sampleRecords(): Array<Record<string, any>> {
  ];
 }

+it("serializes an Arrow Table created in another JavaScript realm", async () => {
+  const context = vm.createContext({
+    TextDecoder,
+    TextEncoder,
+    console,
+    setTimeout,
+    clearTimeout,
+  });
+  vm.runInContext(
+    fs.readFileSync(
+      require.resolve("apache-arrow-15/Arrow.es2015.min"),
+      "utf8",
+    ),
+    context,
+  );
+  const foreignTable: unknown = vm.runInContext(
+    "Arrow.tableFromArrays({ id: new Int32Array([1, 2, 3]), text: ['foo', 'bar', 'baz'] })",
+    context,
+  );
+
+  const foreignMetadata = (
+    foreignTable as { schema: { metadata: Map<string, string> } }
+  ).schema.metadata;
+  expect(foreignMetadata).not.toBeInstanceOf(Map);
+
+  const buf = await fromDataToBuffer(
+    foreignTable as Parameters<typeof fromDataToBuffer>[0],
+  );
+  const actual = currentTableFromIPC(buf);
+
+  expect(actual.numRows).toBe(3);
+  expect(actual.getChild("id")?.toJSON()).toEqual([1, 2, 3]);
+  expect(actual.getChild("text")?.toJSON()).toEqual(["foo", "bar", "baz"]);
+});
+
 it("preserves field metadata from a provided schema", async function () {
  const jsonMetadata = new Map([["ARROW:extension:name", "lance.json"]]);
  const schema = new CurrentSchema([
@@ -72,8 +72,7 @@ export type FieldLike =
    };

 export type DataLike =
-  // biome-ignore lint/suspicious/noExplicitAny: <explanation>
-  | import("apache-arrow").Data<Struct<any>>
+  | import("apache-arrow").Data
  | {
      // biome-ignore lint/suspicious/noExplicitAny: <explanation>
      type: any;
@@ -82,6 +81,7 @@ export type DataLike =
      stride: number;
      nullable: boolean;
      children: DataLike[];
+      dictionary?: { data: readonly DataLike[] };
      get nullCount(): number;
      // biome-ignore lint/suspicious/noExplicitAny: <explanation>
      values: Buffers<any>[BufferType.DATA];
@@ -157,8 +157,8 @@ export {
  TokenizeTableOptions,
  LsmWriteSpec,
  LsmStats,
-  BucketStats,
-  GenerationStats,
+  TableShardStats,
+  SsTableStats,
  MemtableStats,
  ColumnAlteration,
  FieldMetadataUpdate,
@@ -94,17 +94,24 @@ export function sanitizeMetadata(
  if (metadataLike === undefined || metadataLike === null) {
    return undefined;
  }
-  if (!(metadataLike instanceof Map)) {
+
+  let entries: IterableIterator<[unknown, unknown]>;
+  try {
+    entries = Map.prototype.entries.call(metadataLike);
+  } catch {
    throw Error("Expected metadata, if present, to be a Map<string, string>");
  }
-  for (const item of metadataLike) {
-    if (typeof item[0] !== "string" || typeof item[1] !== "string") {
+
+  const metadata = new Map<string, string>();
+  for (const [key, value] of entries) {
+    if (typeof key !== "string" || typeof value !== "string") {
      throw Error(
        "Expected metadata, if present, to be a Map<string, string> but it had non-string keys or values",
      );
    }
+    metadata.set(key, value);
  }
-  return metadataLike as Map<string, string>;
+  return metadata;
 }

 export function sanitizeInt(typeLike: object) {
@@ -55,8 +55,8 @@ import { sanitizeType } from "./sanitize";
 import { IntoSql, toSQL } from "./util";
 export { IndexConfig } from "./native";
 export {
-  BucketStats,
-  GenerationStats,
+  TableShardStats,
+  SsTableStats,
  LsmStats,
  MemtableStats,
 } from "./native";
@@ -741,7 +741,7 @@ export abstract class Table {
   */
  abstract closeLsmWriters(): Promise<void>;
  /**
-   * Seal every bucket's active memtable into a new L0 generation.
+   * Freeze every table shard's active memtable into a new SSTable.
   *
   * 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 L0 → base compaction pass per bucket.
+   * Trigger a background SSTable compaction pass per table shard.
   *
   * 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.
   *
-   * Seals once, then triggers compaction and polls until the L0 that existed
+   * Freezes once, then triggers compaction and polls until the SSTables 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
+   * SSTables 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} includeGenerationRows Also count rows per L0 generation.
+   * @param {boolean} includeSstableRows Also count rows per SSTable.
   *   Off by default because each count opens an uncached Lance dataset.
   * @returns {Promise<LsmStats | undefined>}
   */
  abstract getLsmStats(
-    includeGenerationRows?: boolean,
+    includeSstableRows?: boolean,
  ): Promise<LsmStats | undefined>;
  /** Retrieve the version of the table */

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

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

  async version(): Promise<number> {
@@ -542,11 +542,11 @@ impl Table {
    #[napi(catch_unwind)]
    pub async fn get_lsm_stats(
        &self,
-        include_generation_rows: bool,
+        include_sstable_rows: bool,
    ) -> napi::Result<Option<LsmStats>> {
        let stats = self
            .inner_ref()?
-            .get_lsm_stats(include_generation_rows)
+            .get_lsm_stats(include_sstable_rows)
            .await
            .default_error()?;
        Ok(stats.map(LsmStats::from))
@@ -950,21 +950,21 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
    }
 }

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

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

-/// 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
+/// 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
 /// this endpoint.
 #[napi(object)]
 #[derive(Clone, Debug)]
-pub struct BucketStats {
-    /// The shard this bucket writes.
+pub struct TableShardStats {
+    /// The shard this table shard writes.
    pub shard_id: String,
    /// `"Active"` or `"Sealed"` (drop-table 2PC in flight).
    pub status: String,
@@ -1023,20 +1023,20 @@ pub struct BucketStats {
    /// Highest WAL position the writer has seen. The difference against
    /// `replayAfterWalEntryPosition` is the WAL lag.
    pub wal_entry_position_last_seen: i64,
-    /// 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*
+    /// 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*
    /// 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"` bucket, whose
+    /// Oldest first, active last. Absent for a `"Sealed"` table shard, whose
    /// in-memory state is torn down.
    pub memtables: Option<Vec<MemtableStats>>,
 }

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

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

 impl From<lancedb::table::LsmStats> for LsmStats {
    fn from(stats: lancedb::table::LsmStats) -> Self {
        Self {
-            buckets: stats.buckets.into_iter().map(Into::into).collect(),
+            table_shards: stats.table_shards.into_iter().map(Into::into).collect(),
        }
    }
 }
@@ -270,7 +270,8 @@ def _iter_projection_pairs(
            if isinstance(expr, str):
                yield name, expr
            elif isinstance(expr, Expr):
-                yield name, expr.to_sql()
+                source = expr._column_name()
+                yield name, source if source is not None else expr.to_sql()
        return
    for column in projection:
        if isinstance(column, str):
@@ -280,7 +281,8 @@ def _iter_projection_pairs(
            if isinstance(expr, str):
                yield name, expr
            elif isinstance(expr, Expr):
-                yield name, expr.to_sql()
+                source = expr._column_name()
+                yield name, source if source is not None else expr.to_sql()


 def _set_blob_column(tbl: pa.Table, output_name: str, blobs: pa.Array) -> pa.Table:
@@ -87,6 +87,7 @@ class PyExpr:
    def contains(self, substr: "PyExpr") -> "PyExpr": ...
    def isin(self, values: List["PyExpr"]) -> "PyExpr": ...
    def cast(self, data_type: pa.DataType) -> "PyExpr": ...
+    def column_name(self) -> Optional[str]: ...
    def to_sql(self) -> str: ...

 def expr_col(name: str) -> PyExpr: ...
@@ -384,7 +385,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_generation_rows: bool) -> Optional[dict]: ...
+    async def get_lsm_stats(self, include_sstable_rows: bool) -> Optional[dict]: ...
    async def close_lsm_writers(self) -> None: ...
    @property
    def tags(self) -> Tags: ...
@@ -608,6 +609,7 @@ class PyQueryRequest:
    filter: Optional[Union[str, bytes]]
    full_text_search: Optional[FullTextQuery]
    select: Optional[Union[str, List[str]]]
+    select_source_columns: Optional[Dict[str, str]]
    fast_search: Optional[bool]
    with_row_id: Optional[bool]
    use_lsm: Optional[bool]
@@ -249,6 +249,10 @@ class Expr:

    # ── utilities ────────────────────────────────────────────────────────────

+    def _column_name(self) -> str | None:
+        """Return the source name when this is a bare column expression."""
+        return self._inner.column_name()
+
    def to_sql(self) -> str:
        """Render the expression as a SQL string (useful for debugging)."""
        return self._inner.to_sql()
@@ -312,7 +316,7 @@ def func(name: str, *args: ExprLike) -> Expr:
    --------
    >>> from lancedb.expr import col, func
    >>> func("lower", col("name"))
-    Expr(lower(name))
+    Expr(lower(`name`))
    """
    inner_args = [_coerce(a)._inner for a in args]
    return Expr(expr_func(name, inner_args))
@@ -167,6 +167,12 @@ def _projection_to_scanner_kwargs(columns: QueryProjection) -> Dict[str, Any]:
    return {"columns": projection}


+def _query_request_projection(req: "PyQueryRequest") -> QueryProjection:
+    if req.select_source_columns is not None:
+        return req.select_source_columns
+    return req.select
+
+
 def _scanner_kwargs_for_query(
    query: Query,
    blob_mode: BlobMode,
@@ -2799,15 +2805,16 @@ class AsyncQueryBase(object):

        req = self._inner.to_query_request()
        schema = await self._table.schema()
+        projection = _query_request_projection(req)
        self._blob_auto_row_id = blob_auto_row_id_for_scan(
            schema,
-            req.select,
+            projection,
            with_row_id=self._with_row_id,
        )
        if not self._blob_auto_row_id:
            self._blob_paths = ()
            return
-        self._blob_paths = tuple(blob_v2_projection_sources(schema, req.select).keys())
+        self._blob_paths = tuple(blob_v2_projection_sources(schema, projection).keys())
        self._inner.with_row_id()

    def select(self, columns: Union[List[str], dict[str, str]]) -> Self:
@@ -3894,14 +3901,15 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        blob_paths: tuple[str, ...] = ()
        if self._table is not None:
            schema = await self._table.schema()
+            projection = _query_request_projection(req)
            blob_auto_row_id = blob_auto_row_id_for_scan(
                schema,
-                req.select,
+                projection,
                with_row_id=self._with_row_id,
            )
            if blob_auto_row_id:
                blob_paths = tuple(
-                    blob_v2_projection_sources(schema, req.select).keys()
+                    blob_v2_projection_sources(schema, projection).keys()
                )
        self._blob_auto_row_id = blob_auto_row_id
        self._blob_paths = blob_paths
@@ -36,6 +36,7 @@ from lancedb._lancedb import (
    UpdateResult,
 )
 from lancedb.embeddings.base import EmbeddingFunctionConfig
+from lancedb.expr import Expr
 from lancedb.index import (
    FTS,
    BTree,
@@ -863,7 +864,7 @@ class RemoteTable(Table):

    def update(
        self,
-        where: Optional[str] = None,
+        where: Optional[Union[str, Expr]] = None,
        values: Optional[dict] = None,
        *,
        values_sql: Optional[Dict[str, str]] = None,
@@ -874,9 +875,11 @@ class RemoteTable(Table):

        Parameters
        ----------
-        where: str, optional
-            The SQL where clause to use when updating rows. For example, 'x = 2'
-            or 'x IN (1, 2, 3)'. The filter must not be empty, or it will error.
+        where: str or [Expr][lancedb.expr.Expr], optional
+            The filter condition. Can be a SQL string or a type-safe
+            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
+            [lit][lancedb.expr.lit]. The filter must not be empty, or it will
+            error.
        values: dict, optional
            The values to update. The keys are the column names and the values
            are the values to set.
@@ -1026,11 +1029,11 @@ class RemoteTable(Table):
        [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
        return LOOP.run(self._table.compact_lsm())

-    def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
+    def get_lsm_stats(self, *, include_sstable_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_generation_rows=include_generation_rows)
+            self._table.get_lsm_stats(include_sstable_rows=include_sstable_rows)
        )

    def close_lsm_writers(self) -> None:
@@ -1744,7 +1744,7 @@ class Table(ABC):
    @abstractmethod
    def update(
        self,
-        where: Optional[str] = None,
+        where: Optional[Union[str, Expr]] = None,
        values: Optional[dict] = None,
        *,
        values_sql: Optional[Dict[str, str]] = None,
@@ -1759,9 +1759,11 @@ class Table(ABC):

        Parameters
        ----------
-        where: str, optional
-            The SQL where clause to use when updating rows. For example, 'x = 2'
-            or 'x IN (1, 2, 3)'. The filter must not be empty, or it will error.
+        where: str or [Expr][lancedb.expr.Expr], optional
+            The filter condition. Can be a SQL string or a type-safe
+            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
+            [lit][lancedb.expr.lit]. The filter must not be empty, or it will
+            error.
        values: dict, optional
            The values to update. The keys are the column names and the values
            are the values to set.
@@ -1779,6 +1781,7 @@ class Table(ABC):
        Examples
        --------
        >>> import lancedb
+        >>> from lancedb.expr import col
        >>> import pandas as pd
        >>> data = pd.DataFrame({"x": [1, 2, 3], "vector": [[1.0, 2], [3, 4], [5, 6]]})
        >>> db = lancedb.connect("./.lancedb")
@@ -1788,7 +1791,7 @@ class Table(ABC):
        0  1  [1.0, 2.0]
        1  2  [3.0, 4.0]
        2  3  [5.0, 6.0]
-        >>> table.update(where="x = 2", values={"vector": [10.0, 10]})
+        >>> table.update(where=col("x") == 2, values={"vector": [10.0, 10]})
        UpdateResult(rows_updated=1, version=2)
        >>> table.to_pandas()
           x        vector
@@ -3841,7 +3844,7 @@ class LanceTable(Table):

    def update(
        self,
-        where: Optional[str] = None,
+        where: Optional[Union[str, Expr]] = None,
        values: Optional[dict] = None,
        *,
        values_sql: Optional[Dict[str, str]] = None,
@@ -3852,9 +3855,11 @@ class LanceTable(Table):

        Parameters
        ----------
-        where: str, optional
-            The SQL where clause to use when updating rows. For example, 'x = 2'
-            or 'x IN (1, 2, 3)'. The filter must not be empty, or it will error.
+        where: str or [Expr][lancedb.expr.Expr], optional
+            The filter condition. Can be a SQL string or a type-safe
+            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
+            [lit][lancedb.expr.lit]. The filter must not be empty, or it will
+            error.
        values: dict, optional
            The values to update. The keys are the column names and the values
            are the values to set.
@@ -3872,6 +3877,7 @@ class LanceTable(Table):
        Examples
        --------
        >>> import lancedb
+        >>> from lancedb.expr import col
        >>> import pandas as pd
        >>> data = pd.DataFrame({"x": [1, 2, 3], "vector": [[1.0, 2], [3, 4], [5, 6]]})
        >>> db = lancedb.connect("./.lancedb")
@@ -3881,7 +3887,7 @@ class LanceTable(Table):
        0  1  [1.0, 2.0]
        1  2  [3.0, 4.0]
        2  3  [5.0, 6.0]
-        >>> table.update(where="x = 2", values={"vector": [10.0, 10]})
+        >>> table.update(where=col("x") == 2, values={"vector": [10.0, 10]})
        UpdateResult(rows_updated=1, version=2)
        >>> table.to_pandas()
           x        vector
@@ -4183,11 +4189,11 @@ class LanceTable(Table):
        [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
        return LOOP.run(self._table.compact_lsm())

-    def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
+    def get_lsm_stats(self, *, include_sstable_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_generation_rows=include_generation_rows)
+            self._table.get_lsm_stats(include_sstable_rows=include_sstable_rows)
        )

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

-        One flush, sealing every memtable into L0, then compaction triggers
+        One flush, freezing every memtable into an SSTable, 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: generations created *while* it runs are deliberately not
+        Best-effort: SSTables 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 generations are gone, raises on a terminal server fault, and
+        target SSTables 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
@@ -4930,25 +4936,25 @@ class AsyncTable:
        await self._inner.checkpoint_lsm()

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

-        Does not touch the base table — moving L0 into base is
+        Does not touch the base table — compacting SSTables 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 L0 to base compaction pass per bucket.
+        """Trigger a background SSTable compaction pass per table shard.

        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 L0 has reached base.
+        until the current SSTables have reached base.
        """
        await self._inner.compact_lsm()

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

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

        Parameters
        ----------
-        include_generation_rows
-            Report a row count per L0 generation. Off by default: each count
+        include_sstable_rows
+            Report a row count per SSTable. 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_generation_rows)
+        return await self._inner.get_lsm_stats(include_sstable_rows)

    async def close_lsm_writers(self) -> None:
        """Drain and close any cached MemWAL shard writers for this table.
@@ -5995,7 +6001,7 @@ class AsyncTable:
        self,
        updates: Optional[Dict[str, Any]] = None,
        *,
-        where: Optional[str] = None,
+        where: Optional[Union[str, Expr]] = None,
        updates_sql: Optional[Dict[str, str]] = None,
    ) -> UpdateResult:
        """
@@ -6010,9 +6016,11 @@ class AsyncTable:
            The updates to apply.  The keys should be the name of the column to
            update.  The values should be the new values to assign.  This is
            required unless updates_sql is supplied.
-        where: str, optional
-            An SQL filter that controls which rows are updated. For example, 'x = 2'
-            or 'x IN (1, 2, 3)'.  Only rows that satisfy this filter will be udpated.
+        where: str or [Expr][lancedb.expr.Expr], optional
+            The filter condition. Can be a SQL string or a type-safe
+            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
+            [lit][lancedb.expr.lit]. Only rows that satisfy this filter will
+            be updated.
        updates_sql: dict, optional
            The updates to apply, expressed as SQL expression strings.  The keys should
            be column names. The values should be SQL expressions.  These can be SQL
@@ -6030,13 +6038,14 @@ class AsyncTable:
        --------
        >>> import asyncio
        >>> import lancedb
+        >>> from lancedb.expr import col
        >>> import pandas as pd
        >>> async def demo_update():
        ...     data = pd.DataFrame({"x": [1, 2], "vector": [[1, 2], [3, 4]]})
        ...     db = await lancedb.connect_async("./.lancedb")
        ...     table = await db.create_table("my_table", data)
        ...     # x is [1, 2], vector is [[1, 2], [3, 4]]
-        ...     await table.update({"vector": [10, 10]}, where="x = 2")
+        ...     await table.update({"vector": [10, 10]}, where=col("x") == 2)
        ...     # x is [1, 2], vector is [[1, 2], [10, 10]]
        ...     await table.update(updates_sql={"x": "x + 1"})
        ...     # x is [2, 3], vector is [[1, 2], [10, 10]]
@@ -6050,7 +6059,8 @@ class AsyncTable:
        if updates is not None:
            updates_sql = {k: value_to_sql(v) for k, v in updates.items()}

-        return await self._inner.update(updates_sql, where)
+        predicate = where.to_sql() if isinstance(where, Expr) else where
+        return await self._inner.update(updates_sql, predicate)

    async def add_columns(
        self,
@@ -8,7 +8,12 @@ import pyarrow.compute as pc
 import pytest

 import lancedb
-from lancedb._blob import read_row_ids_from_hits, stash_auto_row_ids
+from lancedb._blob import (
+    blob_v2_projection_sources,
+    read_row_ids_from_hits,
+    stash_auto_row_ids,
+)
+from lancedb.expr import col
 from lancedb.index import FTS
 from lancedb.schema import blob_column_paths, blob_v2_column_paths

@@ -70,6 +75,14 @@ def test_blob_v2_column_paths_include_list_children():
    ]


+def test_blob_v2_projection_sources_use_typed_column_name():
+    schema = pa.schema([lancedb.blob("blob")])
+
+    assert blob_v2_projection_sources(schema, {"blob_alias": col("blob")}) == {
+        "blob_alias": "blob"
+    }
+
+
 def _legacy_v1_table(name):
    db = lancedb.connect("memory:///")
    schema = pa.schema(
@@ -166,6 +179,20 @@ async def test_async_table_to_pandas_descriptions_mode_omits_row_id():
    assert set(descriptor.keys()) == {"kind", "position", "size", "blob_id", "blob_uri"}


+@pytest.mark.asyncio
+async def test_async_typed_blob_projection_preserves_source_column():
+    db = await lancedb.connect_async("memory:///typed_blob_projection")
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("blob")])
+    table = await db.create_table("typed_blob_projection", schema=schema)
+    await table.add([{"id": 1, "blob": b"alpha"}])
+
+    hits = await table.query().select({"blob_alias": col("blob")}).to_arrow()
+
+    assert "_lance_row_id" in hits.schema.field("blob_alias").type.names
+    blobs = await table.fetch_blobs("blob", hits)
+    assert blobs.to_pylist() == [b"alpha"]
+
+
 def test_fetch_blobs_round_trip():
    table = _blob_table(
        "round_trip",
@@ -403,6 +430,50 @@ async def test_blob_v2_hybrid_fetch_blobs_async():
    assert {blobs[i].as_py() for i in range(len(blobs))} == {b"alpha", b"beta"}


+@pytest.mark.asyncio
+async def test_async_hybrid_typed_blob_projection_preserves_source_column():
+    db = await lancedb.connect_async("memory:///hybrid_typed_blob")
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("text", pa.utf8()),
+            pa.field("vector", pa.list_(pa.float32(), list_size=2)),
+            lancedb.blob("blob"),
+        ]
+    )
+    table = await db.create_table("hybrid_typed_blob", schema=schema)
+    await table.add(
+        [
+            {
+                "id": 1,
+                "text": "hello alpha",
+                "vector": [1.0, 0.0],
+                "blob": b"alpha",
+            },
+            {
+                "id": 2,
+                "text": "hello beta",
+                "vector": [0.9, 0.1],
+                "blob": b"beta",
+            },
+        ]
+    )
+    await table.create_index("text", config=FTS(with_position=False))
+
+    hits = await (
+        table.query()
+        .nearest_to([1.0, 0.0])
+        .nearest_to_text("hello")
+        .select({"blob_alias": col("blob")})
+        .limit(2)
+        .to_arrow()
+    )
+
+    assert "_lance_row_id" in hits.schema.field("blob_alias").type.names
+    blobs = await table.fetch_blobs("blob", hits)
+    assert {blobs[i].as_py() for i in range(len(blobs))} == {b"alpha", b"beta"}
+
+
 def test_blob_file_seek_read_and_read_range():
    payload = _identifiable_payload(1024)
    table = _blob_table("seek_read", [{"id": 1, "image": payload}])
@@ -52,7 +52,7 @@ class TestExprConstruction:
    def test_func(self):
        e = func("lower", col("name"))
        assert isinstance(e, Expr)
-        assert e.to_sql() == "lower(name)"
+        assert e.to_sql() == "lower(`name`)"

    def test_func_unknown_raises(self):
        with pytest.raises(Exception):
@@ -115,7 +115,7 @@ class TestExprOperators:
    def test_and_operator(self):
        e = (col("age") > lit(18)) & (col("status") == lit("active"))
        assert isinstance(e, Expr)
-        assert e.to_sql() == "((age > 18) AND (status = 'active'))"
+        assert e.to_sql() == "((age > 18) AND (`status` = 'active'))"

    def test_or_operator(self):
        e = (col("a") == lit(1)) | (col("b") == lit(2))
@@ -166,7 +166,7 @@ class TestExprOperators:
    def test_coerce_plain_str(self):
        e = col("name") == "alice"
        assert isinstance(e, Expr)
-        assert e.to_sql() == "(name = 'alice')"
+        assert e.to_sql() == "(`name` = 'alice')"

    def test_reflexive_comparisons(self):
        # 10 < col("age") swaps to col("age") > 10
@@ -198,85 +198,85 @@ class TestExprBytesLiteral:

    def test_bytes_equality_expr_sql(self):
        e = col("data") == lit(b"\xca\xfe")
-        assert e.to_sql() == "(data = X'CAFE')"
+        assert e.to_sql() == "(`data` = X'CAFE')"

    def test_bytes_ne_expr_sql(self):
        e = col("data") != lit(b"\xff")
-        assert e.to_sql() == "(data <> X'FF')"
+        assert e.to_sql() == "(`data` <> X'FF')"

    def test_bytes_compound_expr_sql(self):
        e = (col("data") == lit(b"\x01")) & (col("id") > lit(5))
-        assert e.to_sql() == "((data = X'01') AND (id > 5))"
+        assert e.to_sql() == "((`data` = X'01') AND (id > 5))"

    def test_bytes_in_function_call(self):
        # Regression test: binary literals inside scalar function calls
        # used to fail because DataFusion's unparser does not support Binary
        # scalars.  Now handled via a placeholder-substitution rewrite.
        e = func("contains", col("data"), lit(b"\xff"))
-        assert e.to_sql() == "contains(data, X'FF')"
+        assert e.to_sql() == "contains(`data`, X'FF')"

    def test_bytes_in_not(self):
        e = ~(col("data") == lit(b"\xff"))
-        assert e.to_sql() == "NOT (data = X'FF')"
+        assert e.to_sql() == "NOT (`data` = X'FF')"


 class TestExprStringMethods:
    def test_lower(self):
        e = col("name").lower()
        assert isinstance(e, Expr)
-        assert e.to_sql() == "lower(name)"
+        assert e.to_sql() == "lower(`name`)"

    def test_upper(self):
        e = col("name").upper()
        assert isinstance(e, Expr)
-        assert e.to_sql() == "upper(name)"
+        assert e.to_sql() == "upper(`name`)"

    def test_contains(self):
        e = col("text").contains(lit("hello"))
        assert isinstance(e, Expr)
-        assert e.to_sql() == "contains(text, 'hello')"
+        assert e.to_sql() == "contains(`text`, 'hello')"

    def test_contains_with_str_coerce(self):
        e = col("text").contains("hello")
        assert isinstance(e, Expr)
-        assert e.to_sql() == "contains(text, 'hello')"
+        assert e.to_sql() == "contains(`text`, 'hello')"

    def test_chained_lower_eq(self):
        e = col("name").lower() == lit("alice")
        assert isinstance(e, Expr)
-        assert e.to_sql() == "(lower(name) = 'alice')"
+        assert e.to_sql() == "(lower(`name`) = 'alice')"


 class TestExprCast:
    def test_cast_string(self):
        e = col("id").cast("string")
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(id AS VARCHAR)"
+        assert e.to_sql() == "arrow_cast(id, 'Utf8')"

    def test_cast_int32(self):
        e = col("score").cast("int32")
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(score AS INTEGER)"
+        assert e.to_sql() == "arrow_cast(score, 'Int32')"

    def test_cast_float64(self):
        e = col("val").cast("float64")
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(val AS DOUBLE)"
+        assert e.to_sql() == "arrow_cast(val, 'Float64')"

    def test_cast_pyarrow_type(self):
        e = col("score").cast(pa.int32())
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(score AS INTEGER)"
+        assert e.to_sql() == "arrow_cast(score, 'Int32')"

    def test_cast_pyarrow_float64(self):
        e = col("val").cast(pa.float64())
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(val AS DOUBLE)"
+        assert e.to_sql() == "arrow_cast(val, 'Float64')"

    def test_cast_pyarrow_string(self):
        e = col("id").cast(pa.string())
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(id AS VARCHAR)"
+        assert e.to_sql() == "arrow_cast(id, 'Utf8')"

    def test_cast_pyarrow_and_string_equivalent(self):
        # pa.int32() and "int32" should produce equivalent SQL
@@ -597,14 +597,14 @@ class TestExprIsin:
    def test_isin_strs(self):
        assert (
            col("status").isin(["active", "pending"]).to_sql()
-            == "status IN ('active', 'pending')"
+            == "`status` IN ('active', 'pending')"
        )

    def test_isin_coerces_and_mixes(self):
        assert col("id").isin([lit(1), 2]).to_sql() == "id IN (1, 2)"

    def test_isin_empty(self):
-        assert col("id").isin([]).to_sql() == "id IN ()"
+        assert col("id").isin([]).to_sql() == "false"

    def test_isin_filter(self, simple_table):
        result = simple_table.search().where(col("id").isin([1, 3, 5])).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_generation_rows": False}]
-        table.get_lsm_stats(include_generation_rows=True)
-        assert seen_bodies[-1] == {"include_generation_rows": True}
+        assert seen_bodies == [{"include_sstable_rows": False}]
+        table.get_lsm_stats(include_sstable_rows=True)
+        assert seen_bodies[-1] == {"include_sstable_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():
-    """Seal, read the watermark, and return once L0 holds nothing.
+    """Freeze, read the watermark, and return once no SSTables remain.

    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 L0 yields no target watermark, so the loop is done
+            # An empty SSTable tier yields no target watermark, so the loop is done
            # after the seal without ever polling compaction.
            send_json(request, {"lsm_stats": {"buckets": []}})
        else:
@@ -11,6 +11,7 @@ import warnings
 import weakref
 from concurrent.futures import ThreadPoolExecutor
 from datetime import date, datetime, timedelta
+from decimal import Decimal
 from time import sleep
 from typing import List
 from unittest.mock import patch
@@ -336,6 +337,21 @@ async def test_update_async(mem_db_async: AsyncConnection):
    assert await table.count_rows("id == 10") == 1


+@pytest.mark.asyncio
+async def test_update_expr_filter_literals_async(mem_db_async: AsyncConnection):
+    values = ["5", "4.66e-84", "it's"]
+    table = await mem_db_async.create_table(
+        "update_expr_literals",
+        data=[{"field": value, "result": "original"} for value in values],
+    )
+
+    for value in values:
+        update_res = await table.update({"result": value}, where=col("field") == value)
+        assert update_res.rows_updated == 1
+
+    assert (await table.to_arrow())["result"].to_pylist() == values
+
+
 def test_create_table(mem_db: DBConnection):
    schema = pa.schema(
        {
@@ -2343,6 +2359,148 @@ def test_update(mem_db: DBConnection):
    assert np.allclose(v, np.array([[1.2, 1.9], [1.1, 1.1]]))


+def test_update_expr_filter_literals(mem_db: DBConnection):
+    values = ["5", "4.66e-84", "it's"]
+    table = mem_db.create_table(
+        "update_expr_literals",
+        data=[{"field": value, "result": "original"} for value in values],
+    )
+
+    for value in values:
+        update_res = table.update(where=col("field") == value, values={"result": value})
+        assert update_res.rows_updated == 1
+
+    assert table.to_arrow()["result"].to_pylist() == values
+
+
+def test_update_expr_filter_preserves_typed_semantics(mem_db: DBConnection):
+    low = Decimal("1.234567890123456789")
+    high = Decimal("1.234567890123456790")
+    decimal_schema = pa.schema(
+        [("val", pa.decimal128(19, 18)), ("result", pa.string())]
+    )
+    decimal_table = mem_db.create_table(
+        "update_expr_decimal",
+        pa.table(
+            {"val": [low, high], "result": ["old", "old"]},
+            schema=decimal_schema,
+        ),
+    )
+    predicate = col("val") < lit(high)
+    assert decimal_table.search().where(predicate).to_arrow().num_rows == 1
+    result = decimal_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    keyword_table = mem_db.create_table(
+        "update_expr_keyword", [{"null": 1, "result": "old"}]
+    )
+    predicate = col("null") == 1
+    assert keyword_table.search().where(predicate).to_arrow().num_rows == 1
+    result = keyword_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    empty_in_table = mem_db.create_table(
+        "update_expr_empty_in", [{"id": 1, "result": "old"}]
+    )
+    predicate = col("id").isin([])
+    assert empty_in_table.search().where(predicate).to_arrow().num_rows == 0
+    result = empty_in_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 0
+
+    marker = "__lancedb_binary_placeholder_0__"
+    binary_schema = pa.schema(
+        [("payload", pa.binary()), ("text", pa.string()), ("result", pa.string())]
+    )
+    binary_table = mem_db.create_table(
+        "update_expr_binary",
+        pa.table(
+            {
+                "payload": [b"\x01", b"\x02"],
+                "text": ["other", marker],
+                "result": ["old", "old"],
+            },
+            schema=binary_schema,
+        ),
+    )
+    predicate = (col("payload") == lit(b"\x01")) | (col("text") == marker)
+    assert binary_table.search().where(predicate).to_arrow().num_rows == 2
+    result = binary_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 2
+
+    nonfinite_table = mem_db.create_table(
+        "update_expr_nonfinite",
+        [{"x": 1.0, "result": "old"}, {"x": 2.0, "result": "old"}],
+    )
+    predicate = col("x") < float("inf")
+    assert nonfinite_table.search().where(predicate).to_arrow().num_rows == 2
+    result = nonfinite_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 2
+
+    float16_table = mem_db.create_table(
+        "update_expr_float16",
+        [{"x": 1.0, "result": "old"}, {"x": 3.0, "result": "old"}],
+    )
+    predicate = col("x").cast(pa.float16()) < 2.0
+    assert float16_table.search().where(predicate).to_arrow().num_rows == 1
+    result = float16_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    string_cast_table = mem_db.create_table(
+        "update_expr_string_cast",
+        [{"x": 1, "result": "old"}, {"x": 2, "result": "old"}],
+    )
+    predicate = col("x").cast("string") == "1"
+    assert string_cast_table.search().where(predicate).to_arrow().num_rows == 1
+    result = string_cast_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    quoted_identifier_schema = pa.schema(
+        [("payload", pa.binary()), ("odd'name", pa.int64()), ("result", pa.string())]
+    )
+    quoted_identifier_table = mem_db.create_table(
+        "update_expr_quoted_identifier",
+        pa.table(
+            {"payload": [b"\x01"], "odd'name": [1], "result": ["old"]},
+            schema=quoted_identifier_schema,
+        ),
+    )
+    predicate = (col("payload") == lit(b"\x01")) & (col("odd'name") == 1)
+    assert quoted_identifier_table.search().where(predicate).to_arrow().num_rows == 1
+    result = quoted_identifier_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    decimal256_schema = pa.schema(
+        [("val", pa.decimal256(40, 2)), ("result", pa.string())]
+    )
+    decimal256_table = mem_db.create_table(
+        "update_expr_decimal256",
+        pa.table(
+            {
+                "val": [Decimal("1.00"), Decimal("3.00")],
+                "result": ["old", "old"],
+            },
+            schema=decimal256_schema,
+        ),
+    )
+    predicate = col("val") < lit(Decimal("2.00")).cast(pa.decimal256(40, 2))
+    assert decimal256_table.search().where(predicate).to_arrow().num_rows == 1
+    result = decimal256_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    binary_empty_table = mem_db.create_table(
+        "update_expr_binary_empty",
+        pa.table(
+            {"payload": [b"\x01", b"\x02"], "result": ["old", "old"]},
+            schema=pa.schema([("payload", pa.binary()), ("result", pa.string())]),
+        ),
+    )
+    predicate = (col("payload") == lit(b"\x01")).isin([])
+    assert binary_empty_table.search().where(predicate).to_arrow().num_rows == 0
+    assert predicate.to_sql() == "false"
+    result = binary_empty_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 0
+
+
 def test_update_with_arrow_scalar(mem_db: DBConnection):
    schema = pa.schema({"id": pa.int64(), "vector": pa.list_(pa.float32(), 4)})
    table = mem_db.create_table("my_table", schema=schema)
@@ -130,6 +130,14 @@ impl PyExpr {

    // ── utilities ────────────────────────────────────────────────────────────

+    /// Return the referenced column name for a bare column expression.
+    fn column_name(&self) -> Option<String> {
+        match &self.0 {
+            DfExpr::Column(column) if column.relation.is_none() => Some(column.name.clone()),
+            _ => None,
+        }
+    }
+
    /// Render the expression as a SQL string (useful for debugging).
    fn to_sql(&self) -> PyResult<String> {
        lancedb::expr::expr_to_sql_string(&self.0).map_err(|e| PyValueError::new_err(e.to_string()))
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;

@@ -325,6 +326,7 @@ pub struct PyQueryRequest {
    pub filter: Option<PyQueryFilter>,
    pub full_text_search: Option<PyLanceDB<FtsQuery>>,
    pub select: PySelect,
+    pub select_source_columns: Option<HashMap<String, String>>,
    pub fast_search: Option<bool>,
    pub with_row_id: Option<bool>,
    pub use_lsm: Option<bool>,
@@ -355,6 +357,7 @@ impl From<AnyQuery> for PyQueryRequest {
                full_text_search: query_request
                    .full_text_search
                    .map(|fts| PyLanceDB(fts.query)),
+                select_source_columns: PySelect::source_columns(&query_request.select),
                select: PySelect(query_request.select),
                fast_search: Some(query_request.fast_search),
                with_row_id: Some(query_request.with_row_id),
@@ -380,6 +383,7 @@ impl From<AnyQuery> for PyQueryRequest {
                offset: vector_query.base.offset,
                filter: vector_query.base.filter.map(PyQueryFilter),
                full_text_search: None,
+                select_source_columns: PySelect::source_columns(&vector_query.base.select),
                select: PySelect(vector_query.base.select),
                fast_search: Some(vector_query.base.fast_search),
                with_row_id: Some(vector_query.base.with_row_id),
@@ -412,6 +416,25 @@ impl From<AnyQuery> for PyQueryRequest {
 #[derive(Clone)]
 pub struct PySelect(Select);

+impl PySelect {
+    fn source_columns(select: &Select) -> Option<HashMap<String, String>> {
+        match select {
+            Select::Expr(pairs) => Some(
+                pairs
+                    .iter()
+                    .filter_map(|(output, expr)| match expr {
+                        lancedb::expr::DfExpr::Column(column) if column.relation.is_none() => {
+                            Some((output.clone(), column.name.clone()))
+                        }
+                        _ => None,
+                    })
+                    .collect(),
+            ),
+            _ => None,
+        }
+    }
+}
+
 impl<'py> IntoPyObject<'py> for PySelect {
    type Target = PyAny;
    type Output = Bound<'py, Self::Target>;
@@ -33,16 +33,16 @@ use pyo3::{

 mod scannable;

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

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

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

@@ -1492,7 +1492,7 @@ impl Table {
        })
    }

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

-    /// Trigger a background L0 → base pass per bucket. Returns once the
+    /// Trigger a background SSTable compaction pass per table shard. 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 +1511,15 @@ impl Table {
    }

    /// Live LSM state, or `None` when the LSM write path is not enabled.
-    #[pyo3(signature = (include_generation_rows=false))]
+    #[pyo3(signature = (include_sstable_rows=false))]
    pub fn get_lsm_stats(
        self_: PyRef<'_, Self>,
-        include_generation_rows: bool,
+        include_sstable_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_generation_rows)
+                .get_lsm_stats(include_sstable_rows)
                .await
                .infer_error()?;
            Python::attach(|py| stats.map(|s| lsm_stats_to_py(py, &s)).transpose())
@@ -157,7 +157,7 @@ mod tests {
        use datafusion_common::ScalarValue;
        let expr = col("data").eq(lit(ScalarValue::Binary(Some(vec![0xca, 0xfe]))));
        let sql = expr_to_sql_string(&expr).unwrap();
-        assert_eq!(sql, "(data = X'CAFE')");
+        assert_eq!(sql, "(`data` = X'CAFE')");
    }

    #[test]
@@ -167,7 +167,7 @@ mod tests {
        let int_expr = col("id").gt(lit(5i64));
        let combined = bin_expr.and(int_expr);
        let sql = expr_to_sql_string(&combined).unwrap();
-        assert_eq!(sql, "((data = X'01') AND (id > 5))");
+        assert_eq!(sql, "((`data` = X'01') AND (id > 5))");
    }

    #[test]
@@ -185,7 +185,7 @@ mod tests {
        // serialized correctly (regression test for placeholder rewrite path).
        let expr = contains(col("data"), lit(ScalarValue::Binary(Some(vec![0xff]))));
        let sql = expr_to_sql_string(&expr).unwrap();
-        assert_eq!(sql, "contains(data, X'FF')");
+        assert_eq!(sql, "contains(`data`, X'FF')");
    }

    #[test]
@@ -196,7 +196,7 @@ mod tests {
            .eq(lit(ScalarValue::Binary(Some(vec![0xab, 0xcd]))))
            .not();
        let sql = expr_to_sql_string(&expr).unwrap();
-        assert_eq!(sql, "NOT (data = X'ABCD')");
+        assert_eq!(sql, "NOT (`data` = X'ABCD')");
    }

    #[test]
@@ -206,6 +206,122 @@ mod tests {
        assert!(sql.contains("IN"), "expected IN in: {}", sql);
    }

+    #[test]
+    fn test_empty_is_in() {
+        let expr = is_in(col("id"), vec![]);
+        assert_eq!(expr_to_sql_string(&expr).unwrap(), "false");
+    }
+
+    #[test]
+    fn test_empty_is_in_discards_binary_children() {
+        use datafusion_common::ScalarValue;
+
+        let expr = is_in(
+            col("payload").eq(lit(ScalarValue::Binary(Some(vec![0x01])))),
+            vec![],
+        );
+        assert_eq!(expr_to_sql_string(&expr).unwrap(), "false");
+    }
+
+    #[test]
+    fn test_keyword_identifier() {
+        let expr = col("null").eq(lit(1i64));
+        assert_eq!(expr_to_sql_string(&expr).unwrap(), "(`null` = 1)");
+    }
+
+    #[test]
+    fn test_decimal_literal_preserves_type() {
+        use datafusion_common::ScalarValue;
+
+        let expr = col("val").lt(lit(ScalarValue::Decimal128(
+            Some(1_234_567_890_123_456_790),
+            19,
+            18,
+        )));
+        let sql = expr_to_sql_string(&expr).unwrap();
+        assert_eq!(
+            sql,
+            "(val < arrow_cast('1.234567890123456790', 'Decimal128(19, 18)'))"
+        );
+    }
+
+    #[test]
+    fn test_non_finite_float_literal_preserves_type() {
+        let expr = col("x").lt(lit(f64::INFINITY));
+        assert_eq!(
+            expr_to_sql_string(&expr).unwrap(),
+            "(x < arrow_cast('inf', 'Float64'))"
+        );
+    }
+
+    #[test]
+    fn test_cast_uses_arrow_type_name() {
+        let string = expr_cast(col("x"), DataType::Utf8);
+        assert_eq!(
+            expr_to_sql_string(&string).unwrap(),
+            "arrow_cast(x, 'Utf8')"
+        );
+
+        let int32 = expr_cast(col("x"), DataType::Int32);
+        assert_eq!(
+            expr_to_sql_string(&int32).unwrap(),
+            "arrow_cast(x, 'Int32')"
+        );
+
+        let expr = expr_cast(col("x"), DataType::Float16).lt(lit(2.0));
+        assert_eq!(
+            expr_to_sql_string(&expr).unwrap(),
+            "(arrow_cast(x, 'Float16') < 2.0)"
+        );
+
+        let decimal = expr_cast(lit("2.00"), DataType::Decimal256(40, 2));
+        assert_eq!(
+            expr_to_sql_string(&decimal).unwrap(),
+            "arrow_cast('2.00', 'Decimal256(40, 2)')"
+        );
+    }
+
+    #[test]
+    fn test_binary_placeholder_does_not_rewrite_user_string() {
+        use datafusion_common::ScalarValue;
+
+        let marker = "__lancedb_binary_placeholder_0__";
+        let expr = col("payload")
+            .eq(lit(ScalarValue::Binary(Some(vec![0x01]))))
+            .or(col("text").eq(lit(marker)));
+        assert_eq!(
+            expr_to_sql_string(&expr).unwrap(),
+            "((payload = X'01') OR (`text` = '__lancedb_binary_placeholder_0__'))"
+        );
+    }
+
+    #[test]
+    fn test_binary_binding_skips_quoted_identifiers() {
+        use datafusion_common::ScalarValue;
+
+        let expr = col("payload")
+            .eq(lit(ScalarValue::Binary(Some(vec![0x01]))))
+            .and(col("odd'name").eq(lit(1i64)))
+            .and(col("odd`'name").eq(lit(2i64)));
+        assert_eq!(
+            expr_to_sql_string(&expr).unwrap(),
+            "(((payload = X'01') AND (`odd'name` = 1)) AND (`odd``'name` = 2))"
+        );
+    }
+
+    #[test]
+    fn test_binary_placeholder_collision_search_is_linear() {
+        use datafusion_common::ScalarValue;
+
+        let collision_shaped = format!("__lancedb_binary_placeholder_0__{}", "_".repeat(64_000));
+        let expr = col("payload")
+            .eq(lit(ScalarValue::Binary(Some(vec![0x01]))))
+            .and(col("text").eq(lit(collision_shaped.clone())));
+        let sql = expr_to_sql_string(&expr).unwrap();
+        assert!(sql.contains("X'01'"));
+        assert!(sql.contains(&format!("'{collision_shaped}'")));
+    }
+
    #[test]
    fn test_multiple_binary_literals() {
        use datafusion_common::ScalarValue;
@@ -1,13 +1,24 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

-use std::any::TypeId;
+use std::{
+    any::TypeId,
+    collections::{HashMap, HashSet},
+};

+use arrow_array::types::{
+    Decimal32Type, Decimal64Type, Decimal128Type, Decimal256Type, DecimalType,
+};
+use arrow_schema::DataType;
 use datafusion_common::ScalarValue;
 use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
 use datafusion_expr::Expr;
+use datafusion_functions::core::expr_fn::{
+    arrow_cast as datafusion_arrow_cast, arrow_try_cast as datafusion_arrow_try_cast,
+};
 use datafusion_sql::sqlparser::{
    dialect::{Dialect as SqlParserDialect, GenericDialect},
+    keywords::ALL_KEYWORDS,
    tokenizer::{Token, Tokenizer},
 };
 use datafusion_sql::unparser::{self, dialect::Dialect as UnparserDialect};
@@ -27,11 +38,13 @@ struct LanceSqlDialect;

 impl UnparserDialect for LanceSqlDialect {
    fn identifier_quote_style(&self, identifier: &str) -> Option<char> {
-        let needs_quote = identifier.chars().any(|c| c.is_ascii_uppercase())
-            || !identifier
-                .chars()
-                .enumerate()
-                .all(|(i, c)| c == '_' || c.is_ascii_alphabetic() || (i > 0 && c.is_ascii_digit()));
+        let identifier_upper = identifier.to_ascii_uppercase();
+        let needs_quote =
+            (identifier_upper != "ID" && ALL_KEYWORDS.contains(&identifier_upper.as_str()))
+                || identifier.chars().any(|c| c.is_ascii_uppercase())
+                || !identifier.chars().enumerate().all(|(i, c)| {
+                    c == '_' || c.is_ascii_alphabetic() || (i > 0 && c.is_ascii_digit())
+                });
        if needs_quote { Some('`') } else { None }
    }
 }
@@ -100,24 +113,128 @@ fn bytes_to_hex_sql(bytes: &[u8]) -> String {
    format!("X'{hex}'")
 }

-/// Returns true if *expr* contains a `Binary` or `LargeBinary` scalar literal
-/// anywhere in its subtree.  DataFusion's SQL unparser cannot serialize those
-/// variants, so we route such expressions through a placeholder-substitution
-/// path that emits SQL `X'...'` byte-string literals.
-fn has_binary_literal(expr: &Expr) -> bool {
-    let mut found = false;
+fn string_literals(expr: &Expr) -> HashSet<String> {
+    let mut literals = HashSet::new();
    let _ = expr.apply(&mut |e: &Expr| {
-        if matches!(
-            e,
-            Expr::Literal(ScalarValue::Binary(_) | ScalarValue::LargeBinary(_), _)
-        ) {
-            found = true;
-            Ok(TreeNodeRecursion::Stop)
-        } else {
-            Ok(TreeNodeRecursion::Continue)
+        if let Expr::Literal(
+            ScalarValue::Utf8(Some(value))
+            | ScalarValue::LargeUtf8(Some(value))
+            | ScalarValue::Utf8View(Some(value)),
+            _,
+        ) = e
+        {
+            literals.insert(value.clone());
        }
+        Ok(TreeNodeRecursion::Continue)
    });
-    found
+    literals
+}
+
+fn typed_string_literal(value: String, data_type: DataType) -> Expr {
+    datafusion_arrow_cast(
+        Expr::Literal(ScalarValue::Utf8(Some(value)), None),
+        Expr::Literal(ScalarValue::Utf8(Some(data_type.to_string())), None),
+    )
+}
+
+fn next_binary_placeholder(user_strings: &HashSet<String>, next_id: &mut usize) -> String {
+    loop {
+        let placeholder = format!("{BINARY_PLACEHOLDER_PREFIX}{}__", *next_id);
+        *next_id += 1;
+        if !user_strings.contains(&placeholder) {
+            return placeholder;
+        }
+    }
+}
+
+fn bind_binary_literals(
+    sql: &str,
+    mut bindings: HashMap<String, Vec<u8>>,
+) -> crate::Result<String> {
+    let bytes = sql.as_bytes();
+    let mut output = Vec::with_capacity(bytes.len());
+    let mut index = 0;
+
+    // Walk SQL string tokens once. Placeholders are plain, unescaped string
+    // literals, so this remains linear even when user strings are large or
+    // deliberately resemble the placeholder prefix.
+    while index < bytes.len() {
+        if bytes[index] == b'`' {
+            let identifier_start = index;
+            index += 1;
+            let mut identifier_end = None;
+            while index < bytes.len() {
+                if bytes[index] == b'`' {
+                    if index + 1 < bytes.len() && bytes[index + 1] == b'`' {
+                        index += 2;
+                    } else {
+                        index += 1;
+                        identifier_end = Some(index);
+                        break;
+                    }
+                } else {
+                    index += 1;
+                }
+            }
+
+            let Some(identifier_end) = identifier_end else {
+                return Err(crate::Error::InvalidInput {
+                    message: "unterminated identifier while binding binary literal".to_string(),
+                });
+            };
+            output.extend_from_slice(&bytes[identifier_start..identifier_end]);
+            continue;
+        }
+
+        if bytes[index] != b'\'' {
+            output.push(bytes[index]);
+            index += 1;
+            continue;
+        }
+
+        let literal_start = index;
+        index += 1;
+        let content_start = index;
+        let mut escaped = false;
+        let mut content_end = None;
+        while index < bytes.len() {
+            if bytes[index] == b'\'' {
+                if index + 1 < bytes.len() && bytes[index + 1] == b'\'' {
+                    escaped = true;
+                    index += 2;
+                } else {
+                    content_end = Some(index);
+                    index += 1;
+                    break;
+                }
+            } else {
+                index += 1;
+            }
+        }
+
+        let Some(content_end) = content_end else {
+            return Err(crate::Error::InvalidInput {
+                message: "unterminated string while binding binary literal".to_string(),
+            });
+        };
+
+        let placeholder = &sql[content_start..content_end];
+        if !escaped && let Some(value) = bindings.remove(placeholder) {
+            output.extend_from_slice(bytes_to_hex_sql(&value).as_bytes());
+        } else {
+            output.extend_from_slice(&bytes[literal_start..index]);
+        }
+    }
+
+    if !bindings.is_empty() {
+        return Err(crate::Error::InvalidInput {
+            message: "failed to bind binary literal while serializing expression".to_string(),
+        });
+    }
+
+    String::from_utf8(output).map_err(|e| crate::Error::InvalidInput {
+        message: format!("failed to bind binary literal: {e}"),
+    })
 }

 fn run_unparser(expr: &Expr) -> crate::Result<String> {
@@ -130,25 +247,37 @@ fn run_unparser(expr: &Expr) -> crate::Result<String> {
 }

 pub fn expr_to_sql_string(expr: &Expr) -> crate::Result<String> {
-    // Fast path: no binary literals — DataFusion's unparser handles everything.
-    if !has_binary_literal(expr) {
-        return run_unparser(expr);
-    }
-
-    // Slow path: DataFusion's unparser cannot serialize `Binary`/`LargeBinary`
-    // scalars, so we rewrite each one to a unique string-literal placeholder,
-    // let the unparser do the rest of the work, then substitute the SQL
-    // `X'...'` byte-string literal back in.  This keeps the operator/function
-    // serialization logic centralized in DataFusion and works for every
-    // expression node type the unparser supports.
-    let mut bindings: Vec<Vec<u8>> = Vec::new();
+    // DataFusion's unparser needs a few adaptations before its SQL can be
+    // reparsed by Lance without changing the typed expression's semantics:
+    //
+    // * decimal literals need an explicit cast to preserve precision and scale;
+    // * casts need exact Arrow type names rather than SQL type aliases;
+    // * an empty IN list is valid in DataFusion but invalid SQL;
+    // * binary literals are unsupported by the unparser and need placeholders.
+    // Eliminate empty membership expressions before visiting their children.
+    // Otherwise a discarded binary child could leave behind a stale binding.
    let rewritten = expr
        .clone()
+        .transform(|e: Expr| match e {
+            Expr::InList(in_list) if in_list.list.is_empty() => Ok(Transformed::yes(
+                Expr::Literal(ScalarValue::Boolean(Some(in_list.negated)), None),
+            )),
+            other => Ok(Transformed::no(other)),
+        })
+        .map_err(|e| crate::Error::InvalidInput {
+            message: format!("failed to rewrite expression: {e}"),
+        })?
+        .data;
+
+    let user_strings = string_literals(&rewritten);
+    let mut next_placeholder_id = 0;
+    let mut binary_bindings = HashMap::new();
+    let rewritten = rewritten
        .transform(|e: Expr| match e {
            Expr::Literal(ScalarValue::Binary(Some(bytes)), m)
            | Expr::Literal(ScalarValue::LargeBinary(Some(bytes)), m) => {
-                let placeholder = format!("{}{}__", BINARY_PLACEHOLDER_PREFIX, bindings.len());
-                bindings.push(bytes);
+                let placeholder = next_binary_placeholder(&user_strings, &mut next_placeholder_id);
+                binary_bindings.insert(placeholder.clone(), bytes);
                Ok(Transformed::yes(Expr::Literal(
                    ScalarValue::Utf8(Some(placeholder)),
                    m,
@@ -158,6 +287,57 @@ pub fn expr_to_sql_string(expr: &Expr) -> crate::Result<String> {
            | Expr::Literal(ScalarValue::LargeBinary(None), m) => {
                Ok(Transformed::yes(Expr::Literal(ScalarValue::Null, m)))
            }
+            Expr::Literal(ScalarValue::Decimal32(Some(value), precision, scale), _m) => {
+                let value = Decimal32Type::format_decimal(value, precision, scale);
+                Ok(Transformed::yes(typed_string_literal(
+                    value,
+                    DataType::Decimal32(precision, scale),
+                )))
+            }
+            Expr::Literal(ScalarValue::Decimal64(Some(value), precision, scale), _m) => {
+                let value = Decimal64Type::format_decimal(value, precision, scale);
+                Ok(Transformed::yes(typed_string_literal(
+                    value,
+                    DataType::Decimal64(precision, scale),
+                )))
+            }
+            Expr::Literal(ScalarValue::Decimal128(Some(value), precision, scale), _m) => {
+                let value = Decimal128Type::format_decimal(value, precision, scale);
+                Ok(Transformed::yes(typed_string_literal(
+                    value,
+                    DataType::Decimal128(precision, scale),
+                )))
+            }
+            Expr::Literal(ScalarValue::Decimal256(Some(value), precision, scale), _m) => {
+                let value = Decimal256Type::format_decimal(value, precision, scale);
+                Ok(Transformed::yes(typed_string_literal(
+                    value,
+                    DataType::Decimal256(precision, scale),
+                )))
+            }
+            Expr::Literal(ScalarValue::Float16(Some(value)), _m) if !value.is_finite() => Ok(
+                Transformed::yes(typed_string_literal(value.to_string(), DataType::Float16)),
+            ),
+            Expr::Literal(ScalarValue::Float32(Some(value)), _m) if !value.is_finite() => Ok(
+                Transformed::yes(typed_string_literal(value.to_string(), DataType::Float32)),
+            ),
+            Expr::Literal(ScalarValue::Float64(Some(value)), _m) if !value.is_finite() => Ok(
+                Transformed::yes(typed_string_literal(value.to_string(), DataType::Float64)),
+            ),
+            Expr::Cast(cast) => Ok(Transformed::yes(datafusion_arrow_cast(
+                *cast.expr,
+                Expr::Literal(
+                    ScalarValue::Utf8(Some(cast.field.data_type().to_string())),
+                    None,
+                ),
+            ))),
+            Expr::TryCast(cast) => Ok(Transformed::yes(datafusion_arrow_try_cast(
+                *cast.expr,
+                Expr::Literal(
+                    ScalarValue::Utf8(Some(cast.field.data_type().to_string())),
+                    None,
+                ),
+            ))),
            other => Ok(Transformed::no(other)),
        })
        .map_err(|e| crate::Error::InvalidInput {
@@ -165,14 +345,12 @@ pub fn expr_to_sql_string(expr: &Expr) -> crate::Result<String> {
        })?
        .data;

-    let mut sql = run_unparser(&rewritten)?;
-    for (i, bytes) in bindings.iter().enumerate() {
-        // The unparser quotes string literals with single quotes, so the
-        // placeholder appears as `'__lancedb_binary_placeholder_<i>__'`.
-        let quoted = format!("'{}{}__'", BINARY_PLACEHOLDER_PREFIX, i);
-        sql = sql.replace(&quoted, &bytes_to_hex_sql(bytes));
+    let sql = run_unparser(&rewritten)?;
+    if binary_bindings.is_empty() {
+        Ok(sql)
+    } else {
+        bind_binary_literals(&sql, binary_bindings)
    }
-    Ok(sql)
 }

 #[cfg(test)]
@@ -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 flushed (L0) generations, deduplicated by primary key
+    /// memtables and the SSTables, 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_generation_rows: bool) -> Result<Option<LsmStats>> {
+    async fn get_lsm_stats(&self, include_sstable_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_generation_rows": include_generation_rows,
+                "include_sstable_rows": include_sstable_rows,
            }));
        let (request_id, response) = self.send_lsm_route(request).await?;
        let body = response.text().await.err_to_http(request_id.clone())?;
@@ -8260,7 +8260,7 @@ mod tests {
        http::Response::builder().status(200).body(body).unwrap()
    }

-    /// A flush landing in an empty L0 finishes on the opening stats read
+    /// A flush landing in an empty SSTable tier 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 "L0 is
+    /// is what lets the loop terminate on a table taking writes where "the SSTable tier 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_generation_rows`
+    /// defaulting to a zero it did not measure. `include_sstable_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_generation_rows"], true,
+                body["include_sstable_rows"], true,
                "the flag must reach the server, not be silently dropped"
            );
            let response = serde_json::json!({
@@ -102,7 +102,7 @@ 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::{BucketStats, GenerationStats, LsmStats, MemtableStats};
+pub use lsm_stats::{LsmStats, MemtableStats, SsTableStats, TableShardStats};
 pub use optimize::{CompactionOptions, OptimizeAction, OptimizeStats};
 pub use refresh::RefreshColumnResult;
 pub use schema_evolution::{
@@ -673,7 +673,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(),
        })
    }
-    /// Seal every bucket's active memtable into L0.
+    /// Freeze every table shard's active memtable into an SSTable.
    ///
    /// The default implementation returns `NotSupported`.
    async fn flush_lsm(&self) -> Result<()> {
@@ -681,7 +681,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 L0 → base compaction pass per bucket.
+    /// Trigger a background SSTable compaction pass per table shard.
    ///
    /// The default implementation returns `NotSupported`.
    async fn compact_lsm(&self) -> Result<()> {
@@ -693,7 +693,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_generation_rows: bool) -> Result<Option<LsmStats>> {
+    async fn get_lsm_stats(&self, _include_sstable_rows: bool) -> Result<Option<LsmStats>> {
        Err(Error::NotSupported {
            message: "get_lsm_stats is not supported on this table type".into(),
        })
@@ -1897,7 +1897,7 @@ impl Table {

    /// Converge this table's LSM write path into its base table.
    ///
-    /// One `flush` to seal every memtable into L0, then compaction triggers
+    /// One `flush` to freeze every memtable into an SSTable, 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 +1932,10 @@ impl Table {
        checkpoint::checkpoint_lsm(self).await
    }

-    /// Seal every bucket's active memtable into L0 without touching the
+    /// Freeze every table shard's active memtable into an SSTable without touching the
    /// base table.
    ///
-    /// Independently useful: flushing makes memtable rows readable from L0 at
+    /// Independently useful: flushing makes memtable rows readable from an SSTable 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 +1943,7 @@ impl Table {
        self.inner.flush_lsm().await
    }

-    /// Run one bounded L0 → base compaction pass per bucket, reporting what
+    /// Run one bounded SSTable compaction pass per table shard, reporting what
    /// it merged and what is left.
    ///
    /// One pass, not convergence: that bounds each request's cost and gives a
@@ -1959,7 +1959,7 @@ impl Table {
    /// state, though on a node that has not claimed this table it claims it,
    /// exactly as a read would.
    ///
-    /// `include_generation_rows` reports a row count per L0 generation. Off by
+    /// `include_sstable_rows` reports a row count per SSTable. Off by
    /// default: each count opens an uncached Lance dataset, and
    /// `checkpoint_lsm` polls this needing only generation numbers.
    ///
@@ -1970,8 +1970,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_generation_rows: bool) -> Result<Option<LsmStats>> {
-        self.inner.get_lsm_stats(include_generation_rows).await
+    pub async fn get_lsm_stats(&self, include_sstable_rows: bool) -> Result<Option<LsmStats>> {
+        self.inner.get_lsm_stats(include_sstable_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 L0 that existed at the start is gone.
+//! generation numbers until the SSTables that existed at the start are 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 L0, then trigger and poll until it drains.
+/// from the resulting SSTables, then trigger and poll until they drain.
 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
-            .buckets
+            .table_shards
            .iter()
-            .filter_map(|b| Some((b.shard_id.clone(), b.newest_generation()?)))
+            .filter_map(|b| Some((b.shard_id.clone(), b.newest_sstable_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.buckets {
+        for b in &stats.table_shards {
            let Some(target) = targets.get(&b.shard_id) else {
                continue;
            };
-            let n = b.outstanding_generations(*target);
+            let n = b.outstanding_sstables(*target);
            if n > 0 {
                outstanding += n;
                all_compacting &= b.compacting;
@@ -1,21 +1,21 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

-//! Live per-bucket LSM state — the shape [`crate::Table::get_lsm_stats`]
+//! Live per-table_shard LSM state — the shape [`crate::Table::get_lsm_stats`]
 //! returns and [`super::checkpoint`] polls.
 //!
-//! Nothing here is derived: sums and differences (total L0 bytes, WAL lag)
+//! Nothing here is derived: sums and differences (total SSTable 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 flushed L0 generation.
+/// One SSTable.
 #[derive(Debug, Clone, Deserialize)]
-pub struct GenerationStats {
+pub struct SsTableStats {
    pub generation: u64,
    pub bytes: u64,
-    /// Present only when `include_generation_rows` was requested. Off by
+    /// Present only when `include_sstable_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 bucket. A table is N buckets on one node; flattening to
-/// a single number hides the one hot bucket that is usually why someone
+/// 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 this endpoint.
 #[derive(Debug, Clone, Deserialize)]
-pub struct BucketStats {
+pub struct TableShardStats {
    pub shard_id: String,
    /// `Active` | `Sealed` (drop-table 2PC in flight).
    pub status: String,
@@ -47,42 +47,42 @@ pub struct BucketStats {
    pub current_generation: u64,
    pub replay_after_wal_entry_position: u64,
    pub wal_entry_position_last_seen: u64,
-    pub generations: Vec<GenerationStats>,
-    /// Whether a pass owns this bucket's compaction latch right now. Says *a*
+    pub sstables: Vec<SsTableStats>,
+    /// Whether a pass owns this table_shard'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` bucket, whose
+    /// Oldest first, active last. Absent for a `Sealed` table_shard, whose
    /// in-memory state is torn down.
    #[serde(default)]
    pub memtables: Option<Vec<MemtableStats>>,
 }

-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()
+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()
    }

-    /// How many generations at or below `target` are still in L0.
+    /// How many SSTables at or below `target` are still uncompacted.
    ///
    /// 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_generations(&self, target: u64) -> usize {
-        self.generations
+    pub(crate) fn outstanding_sstables(&self, target: u64) -> usize {
+        self.sstables
            .iter()
            .filter(|g| g.generation <= target)
            .count()
    }
 }

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

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

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

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

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

    #[test]
-    fn empty_l0_has_no_target() {
-        assert!(bucket("b0", &[], false).newest_generation().is_none());
+    fn an_empty_sstable_tier_has_no_target() {
+        assert!(
+            table_shard("b0", &[], false)
+                .newest_sstable_generation()
+                .is_none()
+        );
    }
 }