raghuvanshraj opened a new issue, #27007:
URL: https://github.com/apache/datafusion/issues/27007
Part of #26517 (Variant epic, section 1). Related: #26282, #25013, #24119,
#20871.
### Is your feature request related to a problem or challenge?
We store Variant data in Parquet. Writers shred the common fields into typed
columns (`typed_value`), as the Parquet Variant shredding spec describes. A
typical query asks for one field:
```sql
-- Spark spelling, also accepted by datafusion-variant
SELECT v:f0::BIGINT FROM t;
-- explicit form
SELECT variant_get(v, 'f0', 'BIGINT') FROM t;
```
Today DataFusion reads the entire Variant column for this query, even though
`f0` is already stored in its own Parquet leaf. The same field stored in a
plain struct column reads only that leaf. On a 265 MB file with 20 shredded
fields, the Variant read is 126.4 MB and the struct read is 11.2 KB for the
same answer.
The reason is that `variant_get` is a scalar UDF, and the Parquet read
planner has no way to learn which leaves of its argument it needs. `get_field`
on a struct is handled specially, so its leaf is projected at the reader. For a
shredded Variant the leaf is right there in the file; nothing reads it.
#### Reproducer
Standalone cargo project, two files. `cargo run --release` writes the data,
runs five queries and prints the `DataSourceExec` line of `EXPLAIN ANALYZE` for
each.
`Cargo.toml`
```toml
[package]
name = "repro-leaf-reads"
version = "0.1.0"
edition = "2021"
# datafusion-variant main pins DataFusion to a main commit on arrow 60.
# We pin the same commit so both crates share one DataFusion.
[dependencies]
datafusion = { git = "https://github.com/apache/datafusion.git";, rev =
"484183dc34ea565190d9cd4a66db6d24789f95d0" }
datafusion-variant = { git =
"https://github.com/datafusion-contrib/datafusion-variant.git";, rev =
"421c9822f2e0500e3696588c410aefad69c61d53" }
arrow = "60"
parquet = { version = "=60.0.0", features = ["variant_experimental"] }
parquet-variant = "=60.0.0"
parquet-variant-compute = "=60.0.0"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
```
<details>
<summary><code>src/main.rs</code></summary>
```rust
//! Reproducer: `variant_get` on a shredded Variant column reads the whole
//! column, while `get_field` on an equivalent struct column reads one leaf.
//!
//! Writes one Parquet file with two columns holding the same 20 fields:
//! s : Struct<f0: Int64, f1..f19: Utf8>
//! v : Variant, shredded with typed_value for all 20 fields
//! then runs the same single-field queries on each and prints the
//! `EXPLAIN ANALYZE` output (projection + scan metrics).
//!
//! Run: cargo run --release
use std::fs::File;
use std::sync::Arc;
use arrow::array::{ArrayRef, Int64Array, RecordBatch, StringArray,
StructArray};
use arrow::datatypes::{DataType, Field, Fields, Schema};
use datafusion::error::Result;
use datafusion::prelude::*;
use parquet::arrow::ArrowWriter;
use parquet::basic::Compression;
use parquet::file::properties::WriterProperties;
use parquet_variant_compute::{ShreddedSchemaBuilder, json_to_variant,
shred_variant};
const ROWS: usize = 100_000;
const ROW_GROUP: usize = 10_000;
const FIELDS: usize = 20;
const STR_LEN: usize = 64;
const PATH: &str = "data.parquet";
fn struct_fields() -> Fields {
let mut f = vec![Field::new("f0", DataType::Int64, true)];
for i in 1..FIELDS {
f.push(Field::new(format!("f{i}"), DataType::Utf8, true));
}
f.into()
}
/// One row group worth of data. f0 is monotonic (row / 100) so that
/// `f0 > 990` selects only the last row group. f1..f19 are 64-byte strings
/// that embed the row number so they do not compress away.
fn chunk(start: usize) -> (ArrayRef, ArrayRef) {
let rows = start..start + ROW_GROUP;
// Struct column
let f0 = Int64Array::from_iter_values(rows.clone().map(|i| (i / 100) as
i64));
let mut cols: Vec<ArrayRef> = vec![Arc::new(f0)];
for k in 1..FIELDS {
let vals: Vec<String> = rows
.clone()
.map(|i| format!("{:0>width$}", i * k, width = STR_LEN))
.collect();
cols.push(Arc::new(StringArray::from(vals)));
}
let s = StructArray::new(struct_fields(), cols, None);
// Variant column: same values, as JSON -> Variant -> shredded Variant
let json: Vec<String> = rows
.clone()
.map(|i| {
let mut parts = vec![format!("\"f0\":{}", i / 100)];
for k in 1..FIELDS {
parts.push(format!(
"\"f{k}\":\"{:0>width$}\"",
i * k,
width = STR_LEN
));
}
format!("{{{}}}", parts.join(","))
})
.collect();
let json: ArrayRef = Arc::new(StringArray::from(json));
let unshredded = json_to_variant(&json).expect("json_to_variant");
let mut shred = ShreddedSchemaBuilder::new()
.with_path("f0", &DataType::Int64)
.expect("path");
for k in 1..FIELDS {
shred = shred
.with_path(format!("f{k}").as_str(), &DataType::Utf8)
.expect("path");
}
let shredded = shred_variant(&unshredded,
&shred.build()).expect("shred_variant");
(Arc::new(s), Arc::new(shredded.into_inner()))
}
fn write_file() -> std::result::Result<(), Box<dyn std::error::Error>> {
let (s0, v0) = chunk(0);
let v_field =
parquet_variant_compute::VariantArray::try_new(v0.as_ref())?.field("v");
let schema = Arc::new(Schema::new(vec![
Field::new("s", s0.data_type().clone(), true),
v_field,
]));
let props = WriterProperties::builder()
.set_compression(Compression::UNCOMPRESSED)
.set_max_row_group_row_count(Some(ROW_GROUP))
.build();
let mut writer = ArrowWriter::try_new(File::create(PATH)?,
schema.clone(), Some(props))?;
for start in (0..ROWS).step_by(ROW_GROUP) {
let (s, v) = if start == 0 { (s0.clone(), v0.clone()) } else {
chunk(start) };
writer.write(&RecordBatch::try_new(schema.clone(), vec![s, v])?)?;
}
writer.close()?;
let bytes = std::fs::metadata(PATH)?.len();
println!("wrote {PATH}: {ROWS} rows, {} row groups, {bytes} bytes", ROWS
/ ROW_GROUP);
// Show the Parquet schema so the shredded layout is visible
let reader =
parquet::file::reader::SerializedFileReader::new(File::open(PATH)?)?;
use parquet::file::reader::FileReader;
let mut out = Vec::new();
parquet::schema::printer::print_schema(&mut out,
reader.metadata().file_metadata().schema());
println!("{}", String::from_utf8_lossy(&out));
for (i, rg) in reader.metadata().row_groups().iter().enumerate().take(1)
{
println!("row group {i}: {} rows, columns:", rg.num_rows());
for c in rg.columns() {
println!(" {:<40} {:>10} bytes", c.column_path().string(),
c.compressed_size());
}
}
Ok(())
}
async fn run(ctx: &SessionContext, label: &str, sql: &str) -> Result<()> {
println!("\n=== {label}\n{sql}");
let rows: usize = ctx.sql(sql).await?.collect().await?.iter().map(|b|
b.num_rows()).sum();
println!("rows returned: {rows}");
let explain = ctx.sql(&format!("EXPLAIN ANALYZE
{sql}")).await?.collect().await?;
let text =
arrow::util::pretty::pretty_format_batches(&explain)?.to_string();
// Print only the DataSourceExec line: projection + scan metrics
for line in text.lines() {
if line.contains("DataSourceExec") {
println!("{}", line.trim_matches(|c| c == '|' || c == ' '));
}
}
Ok(())
}
#[tokio::main]
async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
write_file()?;
// skip_metadata = false keeps the ARROW:extension:name tag on `v`,
// as the datafusion-variant README recommends.
let config = SessionConfig::new()
.set_bool("datafusion.execution.parquet.skip_metadata", false)
.with_target_partitions(1);
let mut ctx = SessionContext::new_with_config(config);
datafusion_variant::register_all(&mut ctx)?;
ctx.register_parquet("t", PATH, ParquetReadOptions::default()).await?;
run(&ctx, "Q1 struct, project one field",
"SELECT s['f0'] FROM t").await?;
run(&ctx, "Q2 variant, project one field",
"SELECT variant_get(v, 'f0', 'BIGINT') FROM t").await?;
run(&ctx, "Q2b variant, same with `:` syntax",
"SELECT v:f0::BIGINT FROM t").await?;
run(&ctx, "Q3 struct, filter on one field",
"SELECT s['f0'] FROM t WHERE s['f0'] > 990").await?;
run(&ctx, "Q4 variant, filter on one field",
"SELECT variant_get(v, 'f0', 'BIGINT') FROM t WHERE variant_get(v,
'f0', 'BIGINT') > 990").await?;
Ok(())
}
```
</details>
The file has two columns with the same 20 fields and the same bytes per leaf:
| Column | Arrow type | `f0` leaf per row group | each `f1..f19` leaf per
row group |
|---|---|---|---|
| `s` | `Struct<f0: Int64, f1..f19: Utf8>` | 1,147 B | 697,572 B |
| `v` | Variant, shredded on all 20 fields (`v.typed_value.fN.typed_value`)
| 1,147 B | 697,572 B |
100,000 rows, 10 row groups, uncompressed, 265,312,566 bytes. `f0 = row /
100`, so `f0 > 990` matches only the last row group.
#### What happens
`DataSourceExec` metrics from `EXPLAIN ANALYZE` (trimmed to the relevant
fields):
| | SQL | projection in scan | predicate in scan | `bytes_scanned` | row
groups with matching rows | row groups read | `elapsed_compute` |
|---|---|---|---|---|---|---|---|
| Q1 | `SELECT s['f0'] FROM t` | `get_field(s@0, f0)` | NA | **11.2 KB** |
10 of 10 | 10 of 10 | 0.62 ms |
| Q2 | `SELECT variant_get(v, 'f0', 'BIGINT') FROM t` | `variant_get(v@1,
f0, BIGINT)` | NA | **126.4 MB** | 10 of 10 | 10 of 10 | 75.1 ms |
| Q2b | `SELECT v:f0::BIGINT FROM t` | `variant_get(v@1, f0)` | NA | **126.4
MB** | 10 of 10 | 10 of 10 | 61.6 ms |
| Q3 | Q1 `WHERE s['f0'] > 990` | `get_field(s@0, f0)` | `get_field(s@0, f0)
> 990` | **11.2 KB** | 1 of 10 (the last) | 10 of 10 | 0.43 ms |
| Q4 | Q2 `WHERE variant_get(v,'f0','BIGINT') > 990` | `v` (whole column) |
none | **126.4 MB** | 1 of 10 (the last) | 10 of 10 | 60.9 ms |
<details>
<summary>Raw <code>DataSourceExec</code> lines for Q1 and Q2</summary>
```
DataSourceExec: ..., projection=[get_field(s@0, f0) as t.s[f0]],
file_type=parquet,
metrics=[output_rows=100.0 K, elapsed_compute=622.08µs, ...,
row_groups_pruned_statistics=10 total → 10 matched, ...,
bytes_scanned=11.2 KB, ..., scan_efficiency_ratio=0% (11.47
K/265.3 M)]
DataSourceExec: ..., projection=[variant_get(v@1, f0, BIGINT) as
variant_get(t.v,Utf8("f0"),Utf8("BIGINT"))], file_type=parquet,
metrics=[output_rows=100.0 K, elapsed_compute=75.11ms, ...,
row_groups_pruned_statistics=10 total → 10 matched, ...,
bytes_scanned=126.4 MB, ..., scan_efficiency_ratio=49.96% (132.6
M/265.3 M)]
```
</details>
Observations:
1. **Projection.** The struct query reads one leaf: 11.2 KB (10 row groups x
1,147 B). The Variant query reads the whole `v` column: 126.4 MB, half the
file. The `f0` leaf exists on the Variant side at the same 1,147 B per row
group; the reader never touches it. Compute time follows bytes: 0.6 ms vs 75 ms.
2. **Spelling does not matter.** `v:f0::BIGINT` plans to the same
`variant_get` call and reads the same bytes as the explicit form.
3. **Filter.** For the struct the predicate is pushed into the scan
(`predicate=get_field(s@0, f0) > 990`), though it prunes no row groups yet (10
of 10 matched; nested-leaf pruning is tracked in #20871). For Variant the
predicate never reaches the scan: `projection=[v]`, no `predicate=`, and the
filter runs in a `FilterExec` above after decoding all 126.4 MB.
#### DataFusion already does this for struct fields
`get_field` on a struct column is understood by the Parquet read planner
(`datafusion/datasource-parquet/src/projection_read_plan.rs`), so `s['f0']`
becomes a projection of the single leaf `s.f0` at the reader. The struct-access
work is tracked in #24119. `variant_get` is an opaque `ScalarUDF`; the planner
can only read its whole input.
A shredded Variant is a struct in the file. The leaf
`v.typed_value.f0.typed_value` is a sibling of `s.f0` with the same size. What
is missing is a way for the UDF to tell the planner which leaves it reads.
### Describe the solution you'd like
A hook by which a `ScalarUDF` declares, per file schema, which leaves of a
nested argument it needs. The Parquet planner then projects those leaves (and
pushes predicates on them) exactly as it does for `get_field`.
Two properties matter for Variant and, we think, for any extension type
built on a struct:
1. **One argument may need more than one leaf.** `variant_get(v, 'f0')` on a
shredded file needs `v.typed_value.f0.typed_value` (the typed column) and
`v.typed_value.f0.value` plus `v.metadata` (the fallback for rows where `f0`
was not shredded, e.g. a different type). The declaration must be a set of
paths, not a single path.
2. **Pruning must respect the fallback.** A row group can be skipped on
`typed_value.f0` statistics only if no row in it has a non-null `f0.value`. A
row with a fallback value has no typed statistic, so pruning on the typed leaf
alone would drop matching rows.
Two open PRs implement a hook of this shape:
- #26282 (`NestedInputAccess`): the UDF answers against each file's schema
at open; includes predicate pruning. Closer to what Variant needs.
- #25013: two `ScalarUDFImpl` methods for the same purpose; now in draft.
Once one lands, `datafusion-variant` implements the hook for `variant_get`.
Nothing Variant-specific is needed in core.
### Describe alternatives you've considered
- **Rewrite in the crate.** Turn `variant_get(v, 'f0')` into
`v['typed_value']['f0']['typed_value']` with an analyzer rule in
`datafusion-variant`. Fast when every row is shredded, wrong otherwise: the
fallback `value` must be read too, and the rewrite cannot see per-file
shredding at plan time.
- **Read the full column and live with it.** This is today. For wide Variant
columns the cost is proportional to the width of all other fields, not to what
the query asks for.
### Additional context
- Part of the Variant epic #26517 (section 1). The hook is generic: the same
mechanism serves any UDF over nested data, e.g. geospatial types stored as
structs.
- Versions used: DataFusion main `484183d` (reports 55.1.0),
`datafusion-variant` main `421c982`, arrow 60.1.0, parquet 60.0.0, rustc 1.98.1.
- Earlier measurement on #25013: 4.3-4.6x end-to-end speedup on 1M rows with
leaf reads.
- Session settings: `datafusion.execution.parquet.skip_metadata = false`
(keeps the `ARROW:extension:name` tag, as the `datafusion-variant` README
recommends), `target_partitions = 1` for a single scan line.
--
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]
