Error with TakeExec and highly nested data

There seems to be an issue with nested data, when retrieving with a filter. This was originally reported upstream: https://github.com/lancedb/lancedb/issues/2217

import lance
import pyarrow as pa

data = pa.Table.from_pylist([
    {
            "item": "test",
            "sub_obj": {
                "second_sub_obj": [
                    {
                        "third_sub_obj": {"int_val": 1},
                        "str_val": "test",
                    }
                ]
            },
        }
])

ds = lance.write_dataset(data, "memory://" )

ds.to_table() # works fine

ds.to_table(filter="item == 'test'") # errors

OSError: Io error: Arrow error: Invalid argument error: column types must match schema types, expected Struct([Field { name: "second_sub_obj", data_type: List(Field { name: "item", data_type: Struct([Field { name: "str_val", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "third_sub_obj", data_type: Struct([Field { name: "int_val", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]) but found Struct([Field { name: "second_sub_obj", data_type: List(Field { name: "item", data_type: Struct([Field { name: "third_sub_obj", data_type: Struct([Field { name: "int_val", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]) at column index 1

basically the difference is:

expected:

struct
  "second_sub_obj": List
    item: struct
      "str_val": string
      "third_sub_obj": struct
        "int_val": int64


found:

struct
  "second_sub_obj": List
    item: struct
      "third_sub_obj": struct
        "int_val": int64

So it's missing "str_val" in there. Which is confusing, because without the filter we find that field in the output.

I think this has to do with V2 storage, since this works fine if we do legacy storage though:

ds = lance.write_dataset(data, "memory://", data_storage_version="legacy" )
ds.to_table(filter="item == 'test'") # fine

pyarrow.Table
item: string
sub_obj: struct<second_sub_obj: list<item: struct<str_val: string, third_sub_obj: struct<int_val: int64>>>>
  child 0, second_sub_obj: list<item: struct<str_val: string, third_sub_obj: struct<int_val: int64>>>
      child 0, item: struct<str_val: string, third_sub_obj: struct<int_val: int64>>
          child 0, str_val: string
          child 1, third_sub_obj: struct<int_val: int64>
              child 0, int_val: int64
----
item: [["test"]]
sub_obj: [
  -- is_valid: all not null
  -- child 0 type: list<item: struct<str_val: string, third_sub_obj: struct<int_val: int64>>>
[      -- is_valid: all not null
      -- child 0 type: string
["test"]
      -- child 1 type: struct<int_val: int64>
        -- is_valid: all not null
        -- child 0 type: int64

Reproduce the nested PyArrow dataset case with lance.write_dataset using default V2 storage, then compare ds.to_table() with and without filter="item == 'test'" and with the legacy storage variant. Trace the filtered read path and nested schema handling; done means the filtered result preserves str_val and matches the unfiltered schema without the reported Arrow error.