Follow-up to #1916 (part of #1827).
#1916 collects per-column null counts on the executor and stores them aggregated on SuccessfulStage.output_column_stats in the scheduler. They are not yet consumed: ShuffleReaderExec::partition_statistics (stats_for_partition in shuffle_reader.rs) still returns column_statistics: vec![ColumnStatistics::new_unknown(); num_fields]
finalise_stage_internal (aqe/planner.rs) — carry output_column_stats alongside partitions.
resolve_shuffle_partitions (aqe/execution_plan/exchange.rs + adaptive.rs) — store stats next to partition locations.
BallistaAdapter::transform_children (aqe/adapter.rs) — pass into ShuffleReaderExec::try_new*.
ShuffleReaderExec — new field plus proto/serde changes (serialized core plan node).
stats_for_partition (shuffle_reader.rs) — use the carried stats instead of new_unknown().
- Collect stats on the sort-based shuffle-write path (
sort_shuffle/writer.rs, currently empty; TODO in code).
- Collect stats on the single-partition write path (
shuffle_writer.rs, currently empty; TODO in code).
Follow-up to #1916 (part of #1827).
#1916 collects per-column null counts on the executor and stores them aggregated on
SuccessfulStage.output_column_statsin the scheduler. They are not yet consumed:ShuffleReaderExec::partition_statistics(stats_for_partitioninshuffle_reader.rs) still returnscolumn_statistics: vec![ColumnStatistics::new_unknown(); num_fields]finalise_stage_internal(aqe/planner.rs) — carryoutput_column_statsalongside partitions.resolve_shuffle_partitions(aqe/execution_plan/exchange.rs+adaptive.rs) — store stats next to partition locations.BallistaAdapter::transform_children(aqe/adapter.rs) — pass intoShuffleReaderExec::try_new*.ShuffleReaderExec— new field plus proto/serde changes (serialized core plan node).stats_for_partition(shuffle_reader.rs) — use the carried stats instead ofnew_unknown().sort_shuffle/writer.rs, currently empty; TODO in code).shuffle_writer.rs, currently empty; TODO in code).