Huge performance regression between 2.0 and 2.1 formats in ordered scan

I have a particular data and query shape, that has a huge regression between the 2.0 and 2.1 on-disk formats when using the default materialization settings.

The data shape looks something like:

• id column (high cardinality, random)
• type column (very low cardinality, in this repro only a single value)
• a decent number of sparse list columns (mostly null), with medium cardinality

Then a query reads the table with ordering by id, while filtering by type (which does not actually filter out any rows, but is needed to trigger the issue).
This performs a lot slower (~80x) with the 2.1 format compared to 2.0.

50000 rows, 52 columns, pylance 8.0.0
format 2.0 late_materialization= None: total   0.05s, first batch  0.031s, median of rest  0.003s
format 2.1 late_materialization= None: total   4.19s, first batch  0.688s, median of rest  0.679s

Forcing early materialization fixes the performance of the query on either version:

50000 rows, 52 columns, pylance 8.0.0
format 2.0 late_materialization=False: total   0.03s, first batch  0.018s, median of rest  0.002s
format 2.1 late_materialization=False: total   0.02s, first batch  0.012s, median of rest  0.002s

It also triggers with sparse string columns instead of sparse lists, to a lesser extent, but still significant (~10x instead of ~80x)

Here is a repro script:

import random
import shutil
import statistics
import time
from pathlib import Path

import lance
import pyarrow as pa

num_rows = 50_000
value_pool = [f"value_{i}" for i in range(1000)]

cols = {
    "id": pa.array([random.randint(0, num_rows) for _ in range(num_rows)]),
    "kind": pa.array(["a"] * num_rows),
}
for c in range(50):
    cols[f"attr_{c:03}"] = pa.array(
        [[random.choice(value_pool)] if random.random() < 0.1 else None for _ in range(num_rows)],
        type=pa.list_(pa.utf8()),
    )
table = pa.table(cols)

for version in ("2.0", "2.1"):
    path = Path(f"/tmp/lance_sorted_scan_{version}")
    if path.exists():
        shutil.rmtree(path)
    lance.write_dataset(table, path, data_storage_version=version, max_rows_per_file=num_rows)

print(f"{num_rows} rows, {len(table.schema)} columns, pylance {lance.__version__}")
for version in ("2.0", "2.1"):
    ds = lance.dataset(f"/tmp/lance_sorted_scan_{version}")
    for late in (False, None):
        scanner = ds.scanner(order_by=["id"], filter="kind = 'a'", batch_size=8192, late_materialization=late)
        t0 = prev = time.perf_counter()
        times = []
        for batch in scanner.to_batches():
            now = time.perf_counter()
            times.append(now - prev)
            prev = now
        print(
            f"format {version} late_materialization={late!s:>5}: total {time.perf_counter() - t0:6.2f}s, "
            f"first batch {times[0]:6.3f}s, median of rest {statistics.median(times[1:]):6.3f}s"
        )