feat[iceberg]: runtime statistics - #676
Conversation
6af5a68 to
63bd029
Compare
|
Seems like CI failure is not related to the current PR scope |
gabotechs
left a comment
gabotechs
left a comment
There was a problem hiding this comment.
Thanks @sandugood! looking very good.
| Ok(Arc::new(Statistics { | ||
| num_rows: Precision::Exact(num_rows), | ||
| total_byte_size: Precision::Exact(total_byte_size), | ||
| column_statistics: vec![], |
There was a problem hiding this comment.
This is the reason why the CI is failing (https://github.com/datafusion-contrib/datafusion-distributed/actions/runs/32901579585/job/97976700594?pr=676).
The schema might declare, for example, 3 columns, but here we are returning a column_statistics vec of 0 columns, and DataFusion, while propagating statistics to upstream nodes, will try to access index 0, 1 and 2 of this empty vec![].
It would be cool if we can actually get some per-column statistics. Do you think that would be possible?
There was a problem hiding this comment.
Fixed by returning ColumnStatistics::new_unknown() for each column of the projected schema. Per-column stats are possible from the manifest entries (null_value_counts, lower_bounds, upper_bounds per data file), but that needs async manifest reads cached at construction, so I'd propose doing it in a follow-up PR.
| let Some(snap) = snapshot else { | ||
| return Ok(Arc::new(Statistics::new_unknown(schema))); | ||
| }; |
There was a problem hiding this comment.
So, if current snapshot is None, we get no statistics. This will be pretty bad, as these stats are essentially what will inform the distributed planner how much to distribute.
Is there any chance of getting the stats from somewhere else that we know it's always going to be present?
There was a problem hiding this comment.
I have changed this piece of code and now populate everything with zeroes in that case.
Why do I think that this is correct? If the table was created, then we would have an entry for that table in the catalog (REST, Glue, HMS etc.). However, if no data files were committed means None for snapshot.
If table wasn't even created we would get error earlier, though.
There was a problem hiding this comment.
🤔 Yeap, your reasoning sounds right. I don't know enough about Iceberg but what you say makes sense, so let's stick to it.
If you happen to have a link to some docs explaining this, cool, otherwise, it's fine to leave it like this.
There was a problem hiding this comment.
https://iceberg.apache.org/spec/#table-metadata-fields
current-snapshot-id is an Optional parameter (for all v1-v4 of Iceberg)
Also I've tried to confirm it with a simple Iceberg table creation with Spark (without any data inserted)
After running DESCRIBE TABLE EXTENDED and checking metadata.json
gabotechs
left a comment
gabotechs
left a comment
There was a problem hiding this comment.
Amazing, great work @sandugood!
| let display = displayable(plan.as_ref()) | ||
| .set_show_statistics(true) |
There was a problem hiding this comment.
Nice! I didn't now this was an option.
|
A follow up to this one might be to have some nice per-column statistics, but that depends more on |
|
Added a comment in the original issue about this. |
2 participants
Tackles #608.
Added runtime statistics that can be extracted from the current snapshot metadata.
total-recordsandtotal-files-sizewhich are being used in thepartition_statisticsfunction.