feat(datafusion): estimate total_byte_size for Lance scans

[vivek-bharathan]

Lance scans reported no byte size, so DataFusion's join guards fell back to row counts and broadcast wide tables. Scan, filtered read and take now report rows times a per-row width: the decoder's estimate of the values plus the Arrow buffers around them, floored at 8 bytes. That floor reproduces the 128 Ki row cap at DataFusion's default threshold ratio.

[westonpace]

I have one blocking question. Otherwise, just some nits about comments. LLMs seem to generate very long and difficult to read comments and it would be nice if we could prune them or make them more concise.

[westonpace]

Blocking: I worry this is too brittle. Variable length fields (string, binary) will report None for byte_width_opt and if I'm reading this correctly that will cause us to report None for the entire schema? We will then return Precision::Absent for estimated_total_byte_size.

Would it be better to at least make some kind of guess in this case?

[vivek-bharathan]

I have another PR in the works to address this issue by measuring the actual data. But now that I think about it, maybe we could start with some default estimate that then gets updated from the cache with the actual measured value. wdyt?

[vivek-bharathan]

Changed: a variable-width column now gets a default width instead of withdrawing the whole row, so total_byte_size is no longer Absent for any schema containing a string. #9477 measures real Arrow bytes from batches a scan has already decoded, caches them per dataset version, and uses them from the next query on. So the first query in a session uses the default and every one after it uses a measurement

[lance-gatekeeper]

❌ Gate recommendation: request changes.

Schema-seeded variable widths reintroduce the merge-insert memory regression. Please keep total_byte_size absent for unmeasured variable-width fields while retaining fixed-width estimates, or derive comparable widths from decoded data before exposing them to DataFusion. Draft #9477 explores measured widths but is stacked on the prior head, so this revision needs safe standalone behavior.

[lance-gatekeeper]

This reintroduces the earlier finding. The 64-byte string seed plus offsets makes a 1,000-row target look smaller than an exact 100-row source, although the target's projected columns occupy 1,056,824 bytes versus 132,136 bytes for the source. DataFusion then builds the hash table from LanceRead. Wider document-like values can make the nominally small target build consume target-sized memory and fail again on retry. Keep total_byte_size absent for unmeasured variable-width fields, or use measured decoded widths; the decoder's batch-sizing seed is not a safe join-size comparison.

Reproducer

Add this test under merge_insert.rs::tests and run cargo test -p lance --lib gate_repro_variable_width_join_orientation -- --nocapture.

#[tokio::test]
async fn gate_repro_variable_width_join_orientation() {
    use arrow_array::{Array, RecordBatchIterator, StringArray, UInt32Array};
    use arrow_schema::{DataType, Field, Schema};

    fn find_hash_join(plan: &dyn ExecutionPlan) -> Option<&HashJoinExec> {
        if let Some(join) = plan.downcast_ref::<HashJoinExec>() {
            return Some(join);
        }
        plan.children()
            .into_iter()
            .find_map(|child| find_hash_join(child.as_ref()))
    }

    let wide = "x".repeat(1_024);
    let target_schema = Arc::new(Schema::new(vec![
        Field::new("key", DataType::UInt32, false),
        Field::new("value", DataType::Utf8, false),
        Field::new("other", DataType::Utf8, true),
    ]));
    let target = RecordBatch::try_new(
        target_schema.clone(),
        vec![
            Arc::new(UInt32Array::from((0..1_000).collect::<Vec<_>>())),
            Arc::new(StringArray::from_iter_values((0..1_000).map(|_| "old"))),
            Arc::new(StringArray::from_iter_values((0..1_000).map(|_| wide.as_str()))),
        ],
    )
    .unwrap();
    let target_bytes =
        target.column(0).get_array_memory_size() + target.column(2).get_array_memory_size();
    let ds = Arc::new(
        Dataset::write(
            RecordBatchIterator::new([Ok(target)], target_schema),
            "memory://",
            None,
        )
        .await
        .unwrap(),
    );

    let source_schema = Arc::new(Schema::new(vec![
        Field::new("key", DataType::UInt32, false),
        Field::new("value", DataType::Utf8, false),
    ]));
    let source = RecordBatch::try_new(
        source_schema.clone(),
        vec![
            Arc::new(UInt32Array::from((0..100).collect::<Vec<_>>())),
            Arc::new(StringArray::from_iter_values((0..100).map(|_| wide.as_str()))),
        ],
    )
    .unwrap();
    let source_bytes: usize = source
        .columns()
        .iter()
        .map(|column| column.get_array_memory_size())
        .sum();
    assert!(target_bytes > source_bytes * 4);

    let provider: Arc<dyn TableProvider> = Arc::new(
        datafusion::datasource::MemTable::try_new(source_schema, vec![vec![source]]).unwrap(),
    );
    let plan = crate::dataset::MergeInsertBuilder::try_new(ds, vec!["key".to_string()])
        .unwrap()
        .when_matched(crate::dataset::WhenMatched::UpdateAll)
        .when_not_matched(crate::dataset::WhenNotMatched::DoNothing)
        .try_build()
        .unwrap()
        .create_plan(provider)
        .await
        .unwrap();
    let join = find_hash_join(plan.as_ref()).expect("a materialized source plans a hash join");
    let build = format!(
        "{}",
        datafusion::physical_plan::displayable(join.left().as_ref()).indent(true)
    );
    assert!(
        build.contains("DataSourceExec") && !build.contains("LanceRead"),
        "source={source_bytes}, target={target_bytes}, build={build}"
    );
}

Expected: the smaller DataSourceExec builds. Observed on 3b5f3a46756e2b1ddc07f8dadb99078859e979f6: the assertion fails with source=132136, target=1056824, build=LanceRead: ... projection=[key, other].

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The variable-width merge-insert regression still reproduces after this rebase. The author confirms that a cold session uses the schema seed; #9477 is now ready but remains stacked on this head and still seeds cold scans, so it does not remove the first-query failure. Please retain fixed-width estimates while leaving total_byte_size absent for unmeasured variable-width output, or derive a comparable data-aware width before publishing it to DataFusion.