`vortex-duckdb` allows DuckDB `Struct` types with duplicate field names

[connortsui20]

Vortex StructFields permits duplicate field names in memory (documented at vortex-array/src/dtype/struct_.rs:122), but DuckDB struct types require unique names: DuckDB's SQL binder rejects duplicates outright, and its Parquet reader silently renames them (a, a becomes a, a_1). The DType to LogicalType conversion in vortex-duckdb does neither.

TryFrom<&StructFields> for LogicalType (vortex-duckdb/src/convert/dtype.rs:318) passes field names through verbatim, and the underlying duckdb_create_struct_type C API does no validation, so a Vortex struct with two identically-named fields produces a DuckDB struct type carrying duplicate names. Name-based field access on that type is then ambiguous.

Minimal reproduction (mirrors the crate's own test_struct_type):

let field_names = FieldNames::from([FieldName::from("a"), FieldName::from("a")]);
let field_types = vec![
    DType::Primitive(PType::I32, Nullability::Nullable),
    DType::Primitive(PType::I64, Nullability::Nullable),
];
let dtype = DType::Struct(StructFields::new(field_names, field_types), Nullability::NonNullable);

// Succeeds, yielding STRUCT(a INTEGER, a BIGINT). DuckDB's binder would reject this
// outright, and its Parquet reader would rename it to STRUCT(a INTEGER, a_1 BIGINT).
let logical_type = LogicalType::try_from(&dtype).unwrap();

The same gap exists on the top-level scan-schema path (extract_schema_from_dtype, vortex-duckdb/src/projection.rs:292) for top-level column names, so scanning a Vortex file whose (possibly nested) schema has duplicate field names reaches the same unchecked conversion.

[connortsui20]

Reproduced this end to end, with one caveat on reachability: the default write strategy already rejects duplicate names at every struct level (StructStrategy bails with StructLayout must have unique field names), so files written the normal way never reach the unchecked conversion. Writing with FlatLayoutStrategy skips that check, and nothing validates field names at read time, so such files can exist:

Details

let inner = StructArray::try_from_iter([
    ("a", buffer![1i32, 2, 3].into_array()),
    ("a", buffer![10i64, 20, 30].into_array()),
])?;
let outer = StructArray::try_from_iter([("s", inner)])?;
SESSION
    .write_options()
    .with_strategy(Arc::new(FlatLayoutStrategy::default()))
    .write(&mut file, outer.into_array().to_array_stream())
    .await?;

Scanning that file through the DuckDB extension misbehaves as predicted. With the duplicates at the top level, DuckDB's binder rejects every query on the file, including COUNT(*), with Binder Error: table "..." has duplicate column name "a". With the duplicates nested in a struct column, the scan binds fine and breaks at execution time instead: SELECT s::VARCHAR fails inside DuckDB with Not implemented Error: Error while casting - duplicate name "a" in struct, and SELECT s.a silently resolves to the first field, so the i64 field is unreachable by name and no error is raised.

Since the writer already blocks duplicates on the default path, the practical exposure is files produced with non-default strategies or by other writers, which points at validating in extract_schema_from_dtype and the StructFields conversion at scan time (or renaming like DuckDB's Parquet reader) rather than at write time.

Here is an e2e regression test (can go in vortex-duckdb/src/e2e_test/vortex_scan_test.rs):

/// A Vortex struct with duplicate field names must not surface to DuckDB as a struct type that
/// carries duplicate child names, because DuckDB requires unique names and misbehaves on types
/// that violate that: casting such a struct to VARCHAR fails at execution time with an internal
/// error ("Not implemented Error: Error while casting - duplicate name \"a\" in struct"), and
/// name-based field access silently resolves to the first matching field, making the other
/// fields unreachable. The scan must instead either reject the bind with a clear error or rename
/// the fields the way DuckDB's own Parquet reader does. See
/// <https://github.com/vortex-data/vortex/issues/8710>.
#[test]
fn test_issue_8710_duplicate_struct_field_names() -> VortexResult<()> {
    let file = RUNTIME.block_on(async {
        // Vortex permits duplicate field names in memory. The default write strategy happens to
        // reject them at every struct level via the StructLayout unique-names check, but the
        // flat layout strategy writes the same file without ever running that check.
        let inner = StructArray::try_from_iter([
            ("a", buffer![1i32, 2, 3].into_array()),
            ("a", buffer![10i64, 20, 30].into_array()),
        ])?;
        let outer = StructArray::try_from_iter([("s", inner)])?;

        let temp_file_path = create_temp_file();
        let mut file = async_fs::File::create(&temp_file_path)
            .await
            .map_err(|e| vortex_err!("{e}"))?;
        SESSION
            .write_options()
            .with_strategy(Arc::new(FlatLayoutStrategy::default()))
            .write(&mut file, outer.into_array().to_array_stream())
            .await?;
        Ok::<_, VortexError>(temp_file_path)
    })?;

    let conn = database_connection();
    let file_path = file.path().to_string_lossy();

    // Rejecting the scan outright at bind time is acceptable behavior.
    if conn.query(&format!("SELECT s FROM '{file_path}'")).is_err() {
        return Ok(());
    }

    // If the scan binds, the struct type handed to DuckDB must be one DuckDB can operate on.
    // With duplicate child names this cast fails inside DuckDB at execution time.
    let result = conn.query(&format!(
        "SELECT string_agg(s::VARCHAR, ';') FROM '{file_path}'"
    ))?;
    let mut chunk = result
        .into_iter()
        .next()
        .ok_or_else(|| vortex_err!("empty result"))?;
    let len: usize = chunk.len().as_();
    let vector = chunk.get_vector_mut(0);
    let aggregated =
        String::from_duckdb_value(&mut unsafe { vector.as_slice_mut::<duckdb_string_t>(len) }[0]);

    // Both fields' values must survive whatever the scan does to disambiguate the names.
    for value in ["10", "20", "30"] {
        assert!(
            aggregated.contains(value),
            "field value {value} is missing from {aggregated:?}"
        );
    }

    Ok(())
}