raghuvanshraj opened a new issue, #27089: URL: https://github.com/apache/datafusion/issues/27089
Part of #26517 (section 3). Related: #26282, #26195, #24109, #25268, #20835, #22908, #16116. ### Is your feature request related to a problem or challenge? A Parquet Variant column can be stored with different shredding layouts in different files of one table: one file writes `Struct<metadata, value>`, another writes `Struct<metadata, value, typed_value: Struct<service, status>>`, a third shreds a different set of fields. Writers do this as the data evolves; Spark reads such tables fine. DataFusion merges the file schemas into one table schema, then the default schema adapter rewrites each file's read expression to match it. A `typed_value` child the file lacks is filled with NULL. For an ordinary struct that is the right thing. For Variant it is not: the shredding spec says a NULL `typed_value` next to a non-null `value` means the value is **not an object**, and readers may skip `value` for any field named in `typed_value` ([VariantShredding.md, Objects](https://github.com/apache/parquet-format/blob/master/VariantShredding.md#objects): "Readers can assume that a value is not an object if `typed_value` is null"). So every shredded field of every row from the unshredded file reads as NULL. No error, no warning. The whole object is still sitting in that row's `value` bytes. Spark spelling of the goal: `SELECT event:service FROM events` returns `service` for every row, whichever layout the row's file used. #### Reproducer Two files per case, written with arrow-rs `shred_variant`. Depends on DataFusion and arrow-rs only; no Variant SQL functions, so the NULL fill is visible with plain struct access. ```toml [package] name = "repro-mixed-files" version = "0.1.0" edition = "2021" [dependencies] datafusion = { git = "https://github.com/apache/datafusion.git";, rev = "1635ed94742c1c060852bc34dc6ebb2302241c7b" } # branch-56 arrow = "60" parquet = { version = "60.1", features = ["variant_experimental"] } parquet-variant-compute = "60.1" tokio = { version = "1", features = ["rt-multi-thread", "macros"] } ``` <details> <summary><code>src/main.rs</code></summary> ```rust use std::fs::{self, File}; use std::sync::Arc; use arrow::array::{ArrayRef, RecordBatch, StringArray}; use arrow::datatypes::{DataType, Schema}; use datafusion::prelude::*; use parquet::arrow::ArrowWriter; use parquet_variant_compute::{json_to_variant, shred_variant, ShreddedSchemaBuilder, VariantArray}; const EVENTS: [&str; 3] = [ r#"{"service":"api","status":200,"latency_ms":12}"#, r#"{"service":"api","status":500,"latency_ms":85}"#, r#"{"service":"db","status":504}"#, ]; fn variant() -> VariantArray { let json: ArrayRef = Arc::new(StringArray::from(EVENTS.to_vec())); json_to_variant(&json).expect("json_to_variant") } fn shredded(paths: &[(&str, DataType)]) -> VariantArray { let mut b = ShreddedSchemaBuilder::new(); for (p, t) in paths { b = b.with_path(*p, t).expect("path"); } shred_variant(&variant(), &b.build()).expect("shred_variant") } fn write(path: &str, v: VariantArray) -> Result<(), Box<dyn std::error::Error>> { let schema = Arc::new(Schema::new(vec![v.field("event")])); let mut w = ArrowWriter::try_new(File::create(path)?, schema.clone(), None)?; w.write(&RecordBatch::try_new(schema, vec![Arc::new(v.into_inner())])?)?; w.close()?; Ok(()) } async fn q(ctx: &SessionContext, sql: &str) -> Result<(), Box<dyn std::error::Error>> { println!("\n{sql}"); match ctx.sql(sql).await?.collect().await { Ok(b) => println!("{}", arrow::util::pretty::pretty_format_batches(&b)?), Err(e) => println!("ERROR: {e}"), } Ok(()) } #[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> { let _ = fs::remove_dir_all("data"); fs::create_dir_all("data/mixed")?; fs::create_dir_all("data/keys")?; // Case 1: one file shredded on service/status, one not shredded at all write("data/mixed/a_shredded.parquet", shredded(&[("service", DataType::Utf8), ("status", DataType::Int64)]))?; write("data/mixed/b_unshredded.parquet", variant())?; // Case 2: two files shredded on different fields write("data/keys/a.parquet", shredded(&[("service", DataType::Utf8), ("status", DataType::Int64)]))?; write("data/keys/b.parquet", shredded(&[("latency_ms", DataType::Int64)]))?; let ctx = SessionContext::new_with_config(SessionConfig::new().with_target_partitions(1)); ctx.register_parquet("mixed", "data/mixed/", ParquetReadOptions::default()).await?; ctx.register_parquet("keys", "data/keys/", ParquetReadOptions::default()).await?; q(&ctx, "SELECT arrow_typeof(event) FROM mixed LIMIT 1").await?; q(&ctx, "SELECT event['typed_value'] IS NULL AS typed_value_is_null, \ octet_length(event['value']) AS value_bytes, \ event['typed_value']['service']['typed_value'] AS service_leaf \ FROM mixed").await?; q(&ctx, "SELECT event['typed_value']['service']['typed_value'] IS NULL AS service_leaf_null, \ event['typed_value']['latency_ms']['typed_value'] IS NULL AS latency_leaf_null, \ octet_length(event['value']) AS value_bytes \ FROM keys").await?; Ok(()) } ``` </details> #### What happens Case 1, one shredded file and one unshredded file. Rows 4-6 come from the unshredded file. The adapter gives them `typed_value = NULL` while `value` holds the whole 18-byte object: ```text SELECT event['typed_value'] IS NULL AS typed_value_is_null, octet_length(event['value']) AS value_bytes, event['typed_value']['service']['typed_value'] AS service_leaf FROM mixed +---------------------+-------------+--------------+ | typed_value_is_null | value_bytes | service_leaf | +---------------------+-------------+--------------+ | false | 7 | api | | false | 7 | api | | false | | db | | true | 18 | | | true | 18 | | | true | 13 | | +---------------------+-------------+--------------+ ``` Case 2, two files shredded on different fields. Each file's rows have NULL in the other file's shredded leaves; the real values are in `value`, where a spec-following reader will not look: ```text SELECT event['typed_value']['service']['typed_value'] IS NULL AS service_leaf_null, event['typed_value']['latency_ms']['typed_value'] IS NULL AS latency_leaf_null, octet_length(event['value']) AS value_bytes FROM keys +-------------------+-------------------+-------------+ | service_leaf_null | latency_leaf_null | value_bytes | +-------------------+-------------------+-------------+ | false | true | 7 | | false | true | 7 | | false | true | | | true | false | 14 | | true | false | 14 | | true | true | 13 | +-------------------+-------------------+-------------+ ``` With `datafusion-variant` (`642248a`) registered, this is what a user sees for case 1. The unshredded file's rows lose `service` and `status`; `latency_ms`, which neither file shreds, is right for all six rows: ```text SELECT variant_get(event, 'service', 'VARCHAR') AS service, event:status::BIGINT AS status, event:latency_ms::BIGINT AS latency_ms FROM mixed +---------+--------+------------+ | service | status | latency_ms | +---------+--------+------------+ | api | 200 | 12 | | api | 500 | 85 | | db | 504 | | | | | 12 | | | | 85 | | | | | +---------+--------+------------+ SELECT count(variant_get(event, 'service', 'VARCHAR')) FROM mixed -> 3 (expected 6) ``` Each file alone reads correctly. #### Where it happens 1. Schema inference merges the files and, since #26195, marks nested children that some files lack as nullable: [`mark_partially_present_fields_nullable`](https://github.com/apache/datafusion/blob/1635ed94742c1c060852bc34dc6ebb2302241c7b/datafusion/datasource-parquet/src/file_format.rs#L277-L333). The merged `typed_value` children are now nullable, so the adapter is allowed to fill them. 2. At open, [`rewrite_column`](https://github.com/apache/datafusion/blob/1635ed94742c1c060852bc34dc6ebb2302241c7b/datafusion/physical-expr-adapter/src/schema_rewriter.rs#L701-L777) in `DefaultPhysicalExprAdapter` sees the file struct differs from the table struct and wraps the column in a `CastExpr`; the arrow struct cast adds the missing child as all-NULL. [`try_rewrite_struct_field_access`](https://github.com/apache/datafusion/blob/1635ed94742c1c060852bc34dc6ebb2302241c7b/datafusion/physical-expr-adapter/src/schema_rewriter.rs#L632) does the same for a direct `event['typed_value']` access on a file that lacks it. 3. Any Variant reader then follows the spec: `typed_value` NULL means not an object, shredded fields are not looked up in `value`. The same adapter-inserted struct cast produced silent wrong results once before, for `get_field` predicates under filter pushdown (#24109, fixed for that path). #26282 anticipates the per-file side of this ("Answers can differ per file, for example when files shred different Variant keys") for the leaves a UDF declares, but states that adapter-inserted struct casts are handled as before, so a read of the whole column, or a `variant_get` without the hook, still goes through the cast. Before #26195 the merged children stayed `non-null` and case 2 failed at open with `Cannot cast struct: target field 'latency_ms' is non-nullable but missing from source`; case 1 already null-filled silently. On `branch-56` both are silent. ### Describe the solution you'd like Null-filling a missing nested child is the right default for ordinary structs (#20835 asked for exactly that), so this is not a request to change the default. The ask is an opt-out for extension types: a file is read in its own layout and reconstructed into the table's logical type by code that knows the type, never null-filled into a physical layout it does not have. Two ways to get there, both generic to extension types: 1. Per-file answers from the UDF hook in #26282: `variant_get` tells the reader which leaves it needs, evaluated against each file's own schema at open. #26282 currently states that adapter-inserted struct casts are handled as before, so it would need to skip the cast for columns the hook claims, or the hook has to see the file struct rather than the cast struct. 2. A hook for a registered extension type to supply its own per-file adaptation (the extension type's adapter decides how a file layout maps to the table type). The Variant adapter would live in `datafusion-variant`; core only needs the extension point. Either way the Variant-specific logic stays out of core, in line with #16116. ### Describe alternatives you've considered - Keep one shredding layout per table. Not under the reader's control; Spark and other writers produce mixed layouts. - Unshred on read into `Struct<metadata, value>` for every file. Works but throws away the shredded columns, so none of the leaf-read work in #26282 / #27007 applies. - Reject mixed layouts at open. Better than silent NULLs, but it fails tables Spark reads fine. ### Additional context DataFusion `branch-56` at `1635ed9`, arrow/parquet 60.1, `datafusion-variant` `642248a`. Run 2026-11-04. The same case 1 result was first seen on DataFusion 54 (0 of 1024 unshredded rows returned). Named earlier on #16116 (2025-07-15) and #22908 (2026-09-08). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
