review fixes

docs/src/js/interfaces/TableStatistics.md

@@ -15,14 +15,19 @@ optional columnBytes: Record<string, number>;
 ```
 
 The compressed on-disk bytes of each column, keyed by dotted field path
-("meta", "meta.geo", "meta.geo.lat"). Every nesting level gets an entry
-covering the field's own bytes plus its whole subtree, so a struct
-reports its total while its children report the breakdown (the
-top-level entries alone sum to `totalBytes`). Counts data-file bytes
-only: blob sidecar payloads and index files are not included, and blob
-columns therefore report just their descriptor bytes. Undefined when
-the backend provides no per-column breakdown (e.g. older remote
-servers).
+("meta", "meta.geo", "meta.geo.lat"). A struct's subfields each get
+their own entry, and every entry covers the field's own bytes plus its
+whole subtree, so a struct reports its total while its children report
+the breakdown (the top-level entries alone sum to `totalBytes`). List
+elements are not broken out: a list column reports a single total with
+its element bytes rolled in. Path segments containing anything other
+than letters, digits, or `_` are backtick-quoted, so a subfield named
+`a.b` under `meta` is keyed as "meta.`a.b`".
+
+Counts data-file bytes only: blob sidecar payloads and index files are
+not included, and blob columns therefore report just their descriptor
+bytes. Undefined when the backend provides no per-column breakdown
+(e.g. older remote servers).
 
 ***
 

nodejs/__test__/table.test.ts

@@ -270,6 +270,23 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
       expect(columnBytes["id"] + columnBytes["meta"]).toBe(stats.totalBytes);
     });
 
+    it("should not break out list elements as their own column", async () => {
+      const conn = await connect(tmpDir.name);
+      const lists = await conn.createTable("list_stats", [
+        { id: 1, tags: ["a", "b"] },
+        { id: 2, tags: ["c", "d"] },
+      ]);
+      const stats = await lists.stats();
+      const columnBytes = stats.columnBytes!;
+
+      // A list's element is an encoding detail, not a column, so there is no
+      // redundant "tags.item" entry duplicating "tags".
+      expect(Object.keys(columnBytes).sort()).toEqual(["id", "tags"]);
+      // The element bytes are rolled into the list column, not dropped.
+      expect(columnBytes["tags"]).toBeGreaterThan(0);
+      expect(columnBytes["id"] + columnBytes["tags"]).toBe(stats.totalBytes);
+    });
+
     it("should overwrite data if asked", async () => {
       const addRes = await table.add([{ id: 1 }, { id: 2 }]);
       expect(addRes).toHaveProperty("version");

nodejs/src/table.rs

@@ -1019,14 +1019,19 @@ pub struct TableStatistics {
     pub fragment_stats: FragmentStatistics,
 
     /// The compressed on-disk bytes of each column, keyed by dotted field path
-    /// ("meta", "meta.geo", "meta.geo.lat"). Every nesting level gets an entry
-    /// covering the field's own bytes plus its whole subtree, so a struct
-    /// reports its total while its children report the breakdown (the
-    /// top-level entries alone sum to `totalBytes`). Counts data-file bytes
-    /// only: blob sidecar payloads and index files are not included, and blob
-    /// columns therefore report just their descriptor bytes. Undefined when
-    /// the backend provides no per-column breakdown (e.g. older remote
-    /// servers).
+    /// ("meta", "meta.geo", "meta.geo.lat"). A struct's subfields each get
+    /// their own entry, and every entry covers the field's own bytes plus its
+    /// whole subtree, so a struct reports its total while its children report
+    /// the breakdown (the top-level entries alone sum to `totalBytes`). List
+    /// elements are not broken out: a list column reports a single total with
+    /// its element bytes rolled in. Path segments containing anything other
+    /// than letters, digits, or `_` are backtick-quoted, so a subfield named
+    /// `a.b` under `meta` is keyed as "meta.`a.b`".
+    ///
+    /// Counts data-file bytes only: blob sidecar payloads and index files are
+    /// not included, and blob columns therefore report just their descriptor
+    /// bytes. Undefined when the backend provides no per-column breakdown
+    /// (e.g. older remote servers).
     pub column_bytes: Option<HashMap<String, i64>>,
 }
 

python/python/lancedb/remote/table.py

@@ -63,7 +63,16 @@ from ..query import (
     LanceTakeQueryBuilder,
     LanceVectorQueryBuilder,
 )
-from ..table import AsyncTable, BlobMode, Branches, IndexStatistics, Query, Table, Tags
+from ..table import (
+    AsyncTable,
+    BlobMode,
+    Branches,
+    IndexStatistics,
+    Query,
+    Table,
+    TableStatistics,
+    Tags,
+)
 from ..types import BaseTokenizerType
 
 
@@ -1010,7 +1019,7 @@ class RemoteTable(Table):
     ):
         return LOOP.run(self._table.wait_for_index(index_names, timeout))
 
-    def stats(self):
+    def stats(self) -> TableStatistics:
         return LOOP.run(self._table.stats())
 
     @property

python/python/lancedb/table.py

@@ -1018,6 +1018,13 @@ class Table(ABC):
     def stats(self) -> TableStatistics:
         """
         Retrieve table and fragment statistics.
+
+        Returns
+        -------
+        TableStatistics
+            Subscript access (``stats["total_bytes"]``) is still supported for
+            backwards compatibility with the older API, which returned a
+            dictionary.
         """
         raise NotImplementedError
 
@@ -4948,8 +4955,15 @@ class AsyncTable:
     async def stats(self) -> TableStatistics:
         """
         Retrieve table and fragment statistics.
+
+        Returns
+        -------
+        TableStatistics
+            Subscript access (``stats["total_bytes"]``) is still supported for
+            backwards compatibility with the older API, which returned a
+            dictionary.
         """
-        return await self._inner.stats()
+        return TableStatistics._from_dict(await self._inner.stats())
 
     async def uri(self) -> str:
         """
@@ -6146,13 +6160,19 @@ class TableStatistics:
         Statistics about fragments in the table.
     column_bytes: dict[str, int] | None
         The compressed on-disk bytes of each column, keyed by dotted field path
-        ("meta", "meta.geo", "meta.geo.lat"). Every nesting level gets an entry
-        covering the field's own bytes plus its whole subtree, so a struct
-        reports its total while its children report the breakdown (the
-        top-level entries alone sum to ``total_bytes``). Counts data-file bytes
-        only: blob sidecar payloads and index files are not included, and blob
-        columns therefore report just their descriptor bytes. ``None`` when the
-        backend provides no per-column breakdown (e.g. older remote servers).
+        ("meta", "meta.geo", "meta.geo.lat"). A struct's subfields each get their
+        own entry, and every entry covers the field's own bytes plus its whole
+        subtree, so a struct reports its total while its children report the
+        breakdown (the top-level entries alone sum to ``total_bytes``). List
+        elements are not broken out: a list column reports a single total with
+        its element bytes rolled in. Path segments containing anything other
+        than letters, digits, or ``_`` are backtick-quoted, so a subfield named
+        ``a.b`` under ``meta`` is keyed as ``meta.`a.b```.
+
+        Counts data-file bytes only: blob sidecar payloads and index files are
+        not included, and blob columns therefore report just their descriptor
+        bytes. ``None`` when the backend provides no per-column breakdown (e.g.
+        older remote servers).
     """
 
     total_bytes: int
@@ -6161,6 +6181,26 @@ class TableStatistics:
     fragment_stats: FragmentStatistics
     column_bytes: Optional[Dict[str, int]] = None
 
+    @classmethod
+    def _from_dict(cls, stats: Dict[str, Any]) -> TableStatistics:
+        fragment_stats = stats["fragment_stats"]
+        return cls(
+            total_bytes=stats["total_bytes"],
+            num_rows=stats["num_rows"],
+            num_indices=stats["num_indices"],
+            fragment_stats=FragmentStatistics(
+                num_fragments=fragment_stats["num_fragments"],
+                num_small_fragments=fragment_stats["num_small_fragments"],
+                lengths=FragmentSummaryStats(**fragment_stats["lengths"]),
+            ),
+            column_bytes=stats.get("column_bytes"),
+        )
+
+    # This exists for backwards compatibility with an older API, which returned
+    # a dictionary instead of a class.
+    def __getitem__(self, key):
+        return getattr(self, key)
+
 
 @dataclass
 class FragmentStatistics:
@@ -6182,6 +6222,11 @@ class FragmentStatistics:
     num_small_fragments: int
     lengths: FragmentSummaryStats
 
+    # This exists for backwards compatibility with an older API, which returned
+    # a dictionary instead of a class.
+    def __getitem__(self, key):
+        return getattr(self, key)
+
 
 @dataclass
 class FragmentSummaryStats:
@@ -6214,6 +6259,11 @@ class FragmentSummaryStats:
     p75: int
     p99: int
 
+    # This exists for backwards compatibility with an older API, which returned
+    # a dictionary instead of a class.
+    def __getitem__(self, key):
+        return getattr(self, key)
+
 
 class Tags:
     """

python/python/tests/test_remote_db.py

@@ -20,6 +20,7 @@ from lancedb.conftest import MockTextEmbeddingFunction
 from lancedb.query import ColumnOrdering
 from lancedb.remote import ClientConfig
 from lancedb.remote.errors import HttpError, RetryError
+from lancedb.table import TableStatistics
 import pytest
 import pyarrow as pa
 
@@ -1009,9 +1010,9 @@ def test_stats():
         table = db.create_table("test", [{"id": 1}])
         res = table.stats()
         print(f"{res=}")
-        # A server that omits the per-column breakdown reports it as None
-        # rather than leaving the key out.
-        assert res == {**stats, "column_bytes": None}
+        # A server that omits the per-column breakdown reports it as None.
+        assert res == TableStatistics._from_dict({**stats, "column_bytes": None})
+        assert res.column_bytes is None
 
 
 def test_stats_column_bytes():
@@ -1053,7 +1054,11 @@ def test_stats_column_bytes():
 
     with mock_lancedb_connection(handler) as db:
         table = db.create_table("test", [{"id": 1}])
-        assert table.stats() == stats
+        res = table.stats()
+        # The dotted paths are opaque to the client: whatever breakdown the
+        # server sends is passed through verbatim.
+        assert res == TableStatistics._from_dict(stats)
+        assert res.column_bytes == stats["column_bytes"]
 
 
 @contextlib.contextmanager

python/python/tests/test_table.py

@@ -24,7 +24,12 @@ from lancedb.db import AsyncConnection, DBConnection
 from lancedb.embeddings import EmbeddingFunctionConfig, EmbeddingFunctionRegistry
 from lancedb.expr import col, lit
 from lancedb.pydantic import LanceModel, Vector
-from lancedb.table import LanceTable
+from lancedb.table import (
+    FragmentStatistics,
+    FragmentSummaryStats,
+    LanceTable,
+    TableStatistics,
+)
 from pydantic import BaseModel
 
 
@@ -3393,26 +3398,25 @@ def test_stats(mem_db: DBConnection):
     assert len(table) == 2
     stats = table.stats()
     print(f"{stats=}")
-    assert stats == {
-        "total_bytes": 60,
-        "num_rows": 2,
-        "num_indices": 0,
-        "fragment_stats": {
-            "num_fragments": 1,
-            "num_small_fragments": 1,
-            "lengths": {
-                "min": 2,
-                "max": 2,
-                "mean": 2,
-                "p25": 2,
-                "p50": 2,
-                "p75": 2,
-                "p99": 2,
-            },
-        },
+    assert stats == TableStatistics(
+        total_bytes=60,
+        num_rows=2,
+        num_indices=0,
+        fragment_stats=FragmentStatistics(
+            num_fragments=1,
+            num_small_fragments=1,
+            lengths=FragmentSummaryStats(
+                min=2, max=2, mean=2, p25=2, p50=2, p75=2, p99=2
+            ),
+        ),
         # Both columns are counted, and the two together make up total_bytes.
-        "column_bytes": {"text": 34, "id": 26},
-    }
+        column_bytes={"text": 34, "id": 26},
+    )
+    # Subscript access still works, for callers written against the older API
+    # that returned a plain dictionary.
+    assert stats["total_bytes"] == stats.total_bytes == 60
+    assert stats["fragment_stats"]["lengths"]["p99"] == 2
+    assert stats["column_bytes"] == stats.column_bytes
 
 
 def test_stats_column_bytes_nested(mem_db: DBConnection):
@@ -3430,7 +3434,8 @@ def test_stats_column_bytes_nested(mem_db: DBConnection):
         data=[{"id": i, "meta": {"a": i, "b": f"value-{i}"}} for i in range(100)],
         schema=schema,
     )
-    column_bytes = table.stats()["column_bytes"]
+    stats = table.stats()
+    column_bytes = stats.column_bytes
 
     # Nested fields get their own dotted-path entries.
     assert set(column_bytes) == {"id", "meta", "meta.a", "meta.b"}
@@ -3439,7 +3444,63 @@ def test_stats_column_bytes_nested(mem_db: DBConnection):
     # A struct's entry covers its whole subtree.
     assert column_bytes["meta"] >= column_bytes["meta.a"] + column_bytes["meta.b"]
     # Top-level entries alone sum to the table total.
-    assert column_bytes["id"] + column_bytes["meta"] == table.stats()["total_bytes"]
+    assert column_bytes["id"] + column_bytes["meta"] == stats.total_bytes
+
+
+def test_stats_column_bytes_list_elements_not_broken_out(mem_db: DBConnection):
+    """A list's element is an encoding detail, not a column, so it must not
+    appear as a redundant ``tags.item`` entry duplicating ``tags``."""
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int32()),
+            pa.field("tags", pa.list_(pa.utf8())),
+            pa.field("vector", pa.list_(pa.float32(), 4)),
+        ]
+    )
+    table = mem_db.create_table(
+        "list_stats",
+        data=[
+            {"id": i, "tags": [f"tag-{i}", "shared"], "vector": [0.1] * 4}
+            for i in range(100)
+        ],
+        schema=schema,
+    )
+    stats = table.stats()
+
+    assert set(stats.column_bytes) == {"id", "tags", "vector"}
+    # The element bytes are rolled into the list column, not dropped.
+    assert stats.column_bytes["tags"] > 0
+    assert sum(stats.column_bytes.values()) == stats.total_bytes
+
+
+def test_stats_column_bytes_dotted_field_name(mem_db: DBConnection):
+    """A subfield whose name contains a dot is backtick-quoted so it stays
+    distinguishable from the nested path that spells the same way."""
+    schema = pa.schema(
+        [
+            pa.field(
+                "meta",
+                pa.struct(
+                    [
+                        pa.field("a.b", pa.int32()),
+                        pa.field("a", pa.struct([pa.field("b", pa.int32())])),
+                    ]
+                ),
+            )
+        ]
+    )
+    table = mem_db.create_table(
+        "dotted_stats",
+        data=[{"meta": {"a.b": i, "a": {"b": i * 1000}}} for i in range(100)],
+        schema=schema,
+    )
+    stats = table.stats()
+
+    # All four fields are present; naive dotted keying would collide the leaf
+    # named "a.b" with the meta -> a -> b path and silently lose one.
+    assert set(stats.column_bytes) == {"meta", "meta.`a.b`", "meta.a", "meta.a.b"}
+    assert stats.column_bytes["meta.a"] == stats.column_bytes["meta.a.b"]
+    assert stats.column_bytes["meta"] == stats.total_bytes
 
 
 @pytest.mark.asyncio
@@ -3449,8 +3510,8 @@ async def test_stats_column_bytes_async(mem_db_async: AsyncConnection):
         data=[{"text": "foo", "id": 0}, {"text": "bar", "id": 1}],
     )
     stats = await table.stats()
-    assert set(stats["column_bytes"]) == {"text", "id"}
-    assert sum(stats["column_bytes"].values()) == stats["total_bytes"]
+    assert set(stats.column_bytes) == {"text", "id"}
+    assert sum(stats.column_bytes.values()) == stats.total_bytes
 
 
 def test_create_table_empty_list_with_schema(mem_db: DBConnection):

rust/lancedb/src/table.rs

@@ -23,6 +23,7 @@ use lance::dataset::{InsertBuilder, WriteParams};
 use lance::index::DatasetIndexExt;
 use lance::index::scalar::load_segment_params;
 use lance::io::{ObjectStoreParams, WrappingObjectStore};
+use lance_core::datatypes::{Field as LanceField, format_field_path};
 use lance_datafusion::utils::StreamingWriteSource;
 use lance_index::IndexCriteria;
 use lance_io::object_store::{LanceNamespaceStorageOptionsProvider, StorageOptionsAccessor};
@@ -3471,38 +3472,17 @@ impl BaseTable for NativeTable {
         let ds_stats = Arc::new(ds_clone).calculate_data_stats().await?;
         let total_bytes = ds_stats.fields.iter().map(|f| f.bytes_on_disk).sum::<u64>() as usize;
 
-        // One entry per field at every nesting level, keyed by dotted path
-        // ("meta", "meta.geo", "meta.geo.lat"). Each entry is the field's own
-        // bytes plus its whole subtree, so a struct reports its total while its
-        // children report the breakdown. Every field gets an entry even when
-        // the stats report no bytes for it (empty table, legacy storage), so
-        // columns appear as 0 rather than being absent.
         let bytes_by_id: HashMap<u32, u64> = ds_stats
             .fields
             .iter()
             .map(|f| (f.id, f.bytes_on_disk))
             .collect();
-        fn subtree_bytes(
-            field: &lance_core::datatypes::Field,
-            path: String,
-            bytes_by_id: &HashMap<u32, u64>,
-            column_bytes: &mut HashMap<String, u64>,
-        ) -> u64 {
-            let mut total = bytes_by_id.get(&(field.id as u32)).copied().unwrap_or(0);
-            for child in &field.children {
-                total += subtree_bytes(
-                    child,
-                    format!("{path}.{}", child.name),
-                    bytes_by_id,
-                    column_bytes,
-                );
-            }
-            column_bytes.insert(path, total);
-            total
-        }
         let mut column_bytes: HashMap<String, u64> = HashMap::new();
+        let mut path = Vec::new();
         for column in &ds.schema().fields {
-            subtree_bytes(column, column.name.clone(), &bytes_by_id, &mut column_bytes);
+            path.push(column.name.as_str());
+            record_column_bytes(column, &mut path, &bytes_by_id, &mut column_bytes);
+            path.pop();
         }
 
         let frags = ds.get_fragments();
@@ -3573,6 +3553,54 @@ impl BaseTable for NativeTable {
     }
 }
 
+/// The bytes stored for `field` itself plus every field nested beneath it, at
+/// any depth.
+///
+/// Fields the stats don't mention count as 0, so a column on an empty table (or
+/// on legacy storage, where `bytes_on_disk` is always 0) reports 0 rather than
+/// being left out of the breakdown entirely.
+fn subtree_bytes(field: &LanceField, bytes_by_id: &HashMap<u32, u64>) -> u64 {
+    let own = u32::try_from(field.id)
+        .ok()
+        .and_then(|id| bytes_by_id.get(&id).copied())
+        .unwrap_or(0);
+    own + field
+        .children
+        .iter()
+        .map(|child| subtree_bytes(child, bytes_by_id))
+        .sum::<u64>()
+}
+
+/// Records one [`TableStatistics::column_bytes`] entry for `field` — keyed by
+/// the dotted path in `path` — and one for each of its named subfields,
+/// recursively.
+///
+/// Only a struct's children are named: a list's element is an encoding detail
+/// with no user-facing name, so its bytes roll up into the list column itself
+/// instead of appearing under a redundant `tags.item` key. Every entry covers
+/// the field's whole subtree either way, so a struct reports its total while
+/// its children report the breakdown.
+///
+/// Paths go through [`format_field_path`], which backtick-quotes any segment
+/// that needs it. Without that, a leaf named `a.b` under `meta` and the path
+/// `meta` -> `a` -> `b` would both key on `meta.a.b` and one would silently
+/// overwrite the other.
+fn record_column_bytes<'a>(
+    field: &'a LanceField,
+    path: &mut Vec<&'a str>,
+    bytes_by_id: &HashMap<u32, u64>,
+    column_bytes: &mut HashMap<String, u64>,
+) {
+    column_bytes.insert(format_field_path(path), subtree_bytes(field, bytes_by_id));
+    if field.logical_type.is_struct() {
+        for child in &field.children {
+            path.push(child.name.as_str());
+            record_column_bytes(child, path, bytes_by_id, column_bytes);
+            path.pop();
+        }
+    }
+}
+
 #[skip_serializing_none]
 #[derive(Debug, Deserialize, PartialEq)]
 pub struct TableStatistics {
@@ -3589,13 +3617,20 @@ pub struct TableStatistics {
     pub fragment_stats: FragmentStatistics,
 
     /// The compressed on-disk bytes of each column, keyed by dotted field
-    /// path ("meta", "meta.geo", "meta.geo.lat"). Every nesting level gets an
-    /// entry covering the field's own bytes plus its whole subtree, so a
-    /// struct reports its total while its children report the breakdown (the
-    /// top-level entries alone sum to `total_bytes`). Counts data-file bytes
-    /// only: blob sidecar payloads and index files are not included, and blob
-    /// columns therefore report just their descriptor bytes. `None` when the
-    /// backend provides no per-column breakdown (e.g. older remote servers).
+    /// path ("meta", "meta.geo", "meta.geo.lat"). A struct's subfields each
+    /// get their own entry, and every entry covers the field's own bytes plus
+    /// its whole subtree, so a struct reports its total while its children
+    /// report the breakdown (the top-level entries alone sum to
+    /// `total_bytes`). List elements are not broken out: a list column
+    /// reports a single total with its element bytes rolled in. Path segments
+    /// containing anything other than letters, digits, or `_` are
+    /// backtick-quoted, so a subfield named `a.b` under `meta` is keyed as
+    /// ``meta.`a.b` ``.
+    ///
+    /// Counts data-file bytes only: blob sidecar payloads and index files are
+    /// not included, and blob columns therefore report just their descriptor
+    /// bytes. `None` when the backend provides no per-column breakdown (e.g.
+    /// older remote servers).
     #[serde(default)]
     pub column_bytes: Option<HashMap<String, u64>>,
 }
@@ -3635,9 +3670,10 @@ mod tests {
     use std::sync::atomic::{AtomicBool, Ordering};
     use std::time::Duration;
 
+    use arrow_array::builder::{ListBuilder, StringBuilder};
     use arrow_array::{
-        ArrayRef, Int32Array, RecordBatch, RecordBatchIterator, RecordBatchReader, StringArray,
-        StructArray,
+        ArrayRef, FixedSizeListArray, Float32Array, Int32Array, RecordBatch, RecordBatchIterator,
+        RecordBatchReader, StringArray, StructArray,
     };
     use arrow_schema::{DataType, Field, Schema};
     use futures::TryStreamExt;
@@ -5209,4 +5245,135 @@ mod tests {
         // Exactly the four expected paths, nothing else.
         assert_eq!(column_bytes.len(), 4);
     }
+
+    /// A list's element child is an encoding detail, not a column, so it must
+    /// not show up as a redundant `tags.item` entry duplicating `tags`.
+    #[tokio::test]
+    pub async fn test_stats_column_bytes_list_elements_not_broken_out() {
+        let tmp_dir = tempdir().unwrap();
+        let uri = tmp_dir.path().to_str().unwrap();
+        let conn = ConnectBuilder::new(uri).execute().await.unwrap();
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new(
+                "tags",
+                DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
+                true,
+            ),
+            Field::new(
+                "vector",
+                DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 4),
+                true,
+            ),
+        ]));
+        let mut tags = ListBuilder::new(StringBuilder::new());
+        for i in 0..100 {
+            tags.values().append_value(format!("tag-{i}"));
+            tags.values().append_value("shared");
+            tags.append(true);
+        }
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Int32Array::from_iter_values(0..100)),
+                Arc::new(tags.finish()),
+                Arc::new(
+                    FixedSizeListArray::try_new(
+                        Arc::new(Field::new("item", DataType::Float32, true)),
+                        4,
+                        Arc::new(Float32Array::from_iter_values(
+                            (0..400).map(|i| i as f32 / 10.0),
+                        )),
+                        None,
+                    )
+                    .unwrap(),
+                ),
+            ],
+        )
+        .unwrap();
+        let table = conn
+            .create_table("list_stats", batch)
+            .execute()
+            .await
+            .unwrap();
+
+        let stats = table.stats().await.unwrap();
+        let column_bytes = stats.column_bytes.unwrap();
+
+        // One entry per column: no `tags.item`, no `vector.item`.
+        assert_eq!(
+            column_bytes.keys().cloned().collect::<HashSet<_>>(),
+            HashSet::from(["id".to_string(), "tags".to_string(), "vector".to_string()])
+        );
+        // The element bytes are rolled into the list column, not dropped.
+        assert!(column_bytes["tags"] > 0);
+        assert!(column_bytes["vector"] > 0);
+        assert_eq!(
+            column_bytes["id"] + column_bytes["tags"] + column_bytes["vector"],
+            stats.total_bytes as u64
+        );
+    }
+
+    /// A subfield whose name contains a dot must not collide with the nested
+    /// path that spells the same way. Here `meta` has both a leaf named `a.b`
+    /// and a struct `a` containing `b`; naive `format!("{path}.{name}")` keying
+    /// would produce `meta.a.b` twice and silently lose one field.
+    #[tokio::test]
+    pub async fn test_stats_column_bytes_dotted_field_name() {
+        let tmp_dir = tempdir().unwrap();
+        let uri = tmp_dir.path().to_str().unwrap();
+        let conn = ConnectBuilder::new(uri).execute().await.unwrap();
+
+        let a_fields = vec![Field::new("b", DataType::Int32, true)];
+        let meta_fields = vec![
+            Field::new("a.b", DataType::Int32, true),
+            Field::new("a", DataType::Struct(a_fields.clone().into()), true),
+        ];
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "meta",
+            DataType::Struct(meta_fields.clone().into()),
+            true,
+        )]));
+        let meta = StructArray::new(
+            meta_fields.into(),
+            vec![
+                Arc::new(Int32Array::from_iter_values(0..100)) as ArrayRef,
+                Arc::new(StructArray::new(
+                    a_fields.into(),
+                    vec![
+                        Arc::new(Int32Array::from_iter_values((0..100).map(|i| i * 1000)))
+                            as ArrayRef,
+                    ],
+                    None,
+                )) as ArrayRef,
+            ],
+            None,
+        );
+        let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(meta)]).unwrap();
+        let table = conn
+            .create_table("dotted_stats", batch)
+            .execute()
+            .await
+            .unwrap();
+
+        let stats = table.stats().await.unwrap();
+        let column_bytes = stats.column_bytes.unwrap();
+
+        // All four fields are present, with the dotted name backtick-quoted so
+        // it stays distinguishable from the `meta` -> `a` -> `b` path.
+        assert_eq!(
+            column_bytes.keys().cloned().collect::<HashSet<_>>(),
+            HashSet::from([
+                "meta".to_string(),
+                "meta.`a.b`".to_string(),
+                "meta.a".to_string(),
+                "meta.a.b".to_string(),
+            ])
+        );
+        assert!(column_bytes["meta.`a.b`"] > 0);
+        assert!(column_bytes["meta.a.b"] > 0);
+        assert_eq!(column_bytes["meta.a"], column_bytes["meta.a.b"]);
+        assert_eq!(column_bytes["meta"], stats.total_bytes as u64);
+    }
 }