put this in the APIs too: python, ts, rust

python/python/lancedb/table.py

@@ -6144,12 +6144,22 @@ class TableStatistics:
         The total number of indices in the table.
     fragment_stats: FragmentStatistics
         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).
     """
 
     total_bytes: int
     num_rows: int
     num_indices: int
     fragment_stats: FragmentStatistics
+    column_bytes: Optional[Dict[str, int]] = None
 
 
 @dataclass

python/python/tests/test_remote_db.py

@@ -1009,7 +1009,51 @@ def test_stats():
         table = db.create_table("test", [{"id": 1}])
         res = table.stats()
         print(f"{res=}")
-        assert res == stats
+        # A server that omits the per-column breakdown reports it as None
+        # rather than leaving the key out.
+        assert res == {**stats, "column_bytes": None}
+
+
+def test_stats_column_bytes():
+    stats = {
+        "total_bytes": 38,
+        "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,
+            },
+        },
+        "column_bytes": {"id": 20, "meta": 18, "meta.a": 18},
+    }
+
+    def handler(request):
+        if request.path == "/v1/table/test/create/?mode=create":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b"{}")
+        elif request.path == "/v1/table/test/stats/":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(json.dumps(stats).encode())
+        else:
+            print(request.path)
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        table = db.create_table("test", [{"id": 1}])
+        assert table.stats() == stats
 
 
 @contextlib.contextmanager

python/python/tests/test_table.py

@@ -3410,9 +3410,49 @@ def test_stats(mem_db: DBConnection):
                 "p99": 2,
             },
         },
+        # Both columns are counted, and the two together make up total_bytes.
+        "column_bytes": {"text": 34, "id": 26},
     }
 
 
+def test_stats_column_bytes_nested(mem_db: DBConnection):
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int32()),
+            pa.field(
+                "meta",
+                pa.struct([pa.field("a", pa.int32()), pa.field("b", pa.utf8())]),
+            ),
+        ]
+    )
+    table = mem_db.create_table(
+        "nested_stats",
+        data=[{"id": i, "meta": {"a": i, "b": f"value-{i}"}} for i in range(100)],
+        schema=schema,
+    )
+    column_bytes = table.stats()["column_bytes"]
+
+    # Nested fields get their own dotted-path entries.
+    assert set(column_bytes) == {"id", "meta", "meta.a", "meta.b"}
+    assert column_bytes["meta.a"] > 0
+    assert column_bytes["meta.b"] > 0
+    # 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"]
+
+
+@pytest.mark.asyncio
+async def test_stats_column_bytes_async(mem_db_async: AsyncConnection):
+    table = await mem_db_async.create_table(
+        "my_table",
+        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"]
+
+
 def test_create_table_empty_list_with_schema(mem_db: DBConnection):
     """Test creating table with empty list data and schema
 

python/src/table.rs

@@ -944,6 +944,10 @@ impl Table {
                 fragment_stats.set_item("lengths", fragment_lengths)?;
                 dict.set_item("fragment_stats", fragment_stats)?;
 
+                // `None` when the backend reports no per-column breakdown
+                // (e.g. an older remote server).
+                dict.set_item("column_bytes", stats.column_bytes)?;
+
                 Ok(Some(dict.unbind()))
             })
         })