Add a standard compressed serialized representation of plans with metrics

[gabotechs]

When troubleshooting performance issues, the plans we render today with display_plan_ascii are the best tool for taking a look at what happened during the query. A displayed plan looks like this:

┌───── DistributedExec ── Tasks: t0:[p0]
│ SortPreservingMergeExec: [l_returnflag@0 ASC NULLS LAST, l_linestatus@1 ASC NULLS LAST]
│   [Stage 2] => NetworkCoalesceExec: output_partitions=6, input_tasks=2
└──────────────────────────────────────────────────
  ┌───── Stage 2 ── Tasks: t0:[p0..p2] t1:[p0..p2]
  │ SortExec: expr=[l_returnflag@0 ASC NULLS LAST, l_linestatus@1 ASC NULLS LAST], preserve_partitioning=[true]
  │   ProjectionExec: expr=[l_returnflag@0 as l_returnflag, l_linestatus@1 as l_linestatus, sum(lineitem.l_quantity)@2 as sum_qty, sum(lineitem.l_extendedprice)@3 as sum_base_price, sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@4 as sum_disc_price, sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) + lineitem.l_tax)@5 as sum_charge, avg(lineitem.l_quantity)@6 as avg_qty, avg(lineitem.l_extendedprice)@7 as avg_price, avg(lineitem.l_discount)@8 as avg_disc, count(Int64(1))@9 as count_order]
  │     AggregateExec: mode=FinalPartitioned, gby=[l_returnflag@0 as l_returnflag, l_linestatus@1 as l_linestatus], aggr=[sum(lineitem.l_quantity), sum(lineitem.l_extendedprice), sum(__common_expr_1) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount), sum(__common_expr_1 * Some(1),20,0 + lineitem.l_tax) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) + lineitem.l_tax), avg(lineitem.l_quantity), avg(lineitem.l_extendedprice), avg(lineitem.l_discount), count(Int64(1))]
  │       [Stage 1] => NetworkShuffleExec: output_partitions=3, input_tasks=4
  └──────────────────────────────────────────────────
    ┌───── Stage 1 ── Tasks: t0:[p0..p5] t1:[p0..p5] t2:[p0..p5] t3:[p0..p5]
    │ RepartitionExec: partitioning=Hash([l_returnflag@0, l_linestatus@1], 6), input_partitions=3
    │   AggregateExec: mode=Partial, gby=[l_returnflag@5 as l_returnflag, l_linestatus@6 as l_linestatus], aggr=[sum(lineitem.l_quantity), sum(lineitem.l_extendedprice), sum(__common_expr_1) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount), sum(__common_expr_1 * Some(1),20,0 + lineitem.l_tax) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) + lineitem.l_tax), avg(lineitem.l_quantity), avg(lineitem.l_extendedprice), avg(lineitem.l_discount), count(Int64(1))]
    │     ProjectionExec: expr=[l_extendedprice@0 * (Some(1),20,0 - l_discount@1) as __common_expr_1, l_quantity@2 as l_quantity, l_extendedprice@0 as l_extendedprice, l_discount@1 as l_discount, l_tax@3 as l_tax, l_returnflag@4 as l_returnflag, l_linestatus@5 as l_linestatus]
    │       FilterExec: l_shipdate@6 <= 1998-09-02, projection=[l_extendedprice@1, l_discount@2, l_quantity@0, l_tax@3, l_returnflag@4, l_linestatus@5]
    │         DistributedLeafExec:
    │           t0: DataSourceExec: file_groups={3 groups: [[/testdata/tpch/plan_sf0.02/lineitem/1.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/10.parquet:<int>..<int>], [/testdata/tpch/plan_sf0.02/lineitem/14.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/15.parquet:<int>..<int>], [/testdata/tpch/plan_sf0.02/lineitem/4.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/5.parquet:<int>..<int>]]}, projection=[l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate], file_type=parquet, predicate=l_shipdate@10 <= 1998-09-02, pruning_predicate=l_shipdate_null_count@1 != row_count@2 AND l_shipdate_min@0 <= 1998-09-02, required_guarantees=[]
    │           t1: DataSourceExec: file_groups={3 groups: [[/testdata/tpch/plan_sf0.02/lineitem/10.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/11.parquet:<int>..<int>], [/testdata/tpch/plan_sf0.02/lineitem/15.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/16.parquet:<int>..<int>], [/testdata/tpch/plan_sf0.02/lineitem/5.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/6.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/7.parquet:<int>..<int>]]}, projection=[l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate], file_type=parquet, predicate=l_shipdate@10 <= 1998-09-02, pruning_predicate=l_shipdate_null_count@1 != row_count@2 AND l_shipdate_min@0 <= 1998-09-02, required_guarantees=[]
    │           t2: DataSourceExec: file_groups={3 groups: [[/testdata/tpch/plan_sf0.02/lineitem/11.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/12.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/13.parquet:<int>..<int>], [/testdata/tpch/plan_sf0.02/lineitem/16.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/2.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/3.parquet:<int>..<int>], [/testdata/tpch/plan_sf0.02/lineitem/7.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/8.parquet:<int>..<int>]]}, projection=[l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate], file_type=parquet, predicate=l_shipdate@10 <= 1998-09-02, pruning_predicate=l_shipdate_null_count@1 != row_count@2 AND l_shipdate_min@0 <= 1998-09-02, required_guarantees=[]
    │           t3: DataSourceExec: file_groups={3 groups: [[/testdata/tpch/plan_sf0.02/lineitem/13.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/14.parquet:<int>..<int>], [/testdata/tpch/plan_sf0.02/lineitem/3.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/4.parquet:<int>..<int>], [/testdata/tpch/plan_sf0.02/lineitem/8.parquet:<int>..<int>, /testdata/tpch/plan_sf0.02/lineitem/9.parquet:<int>..<int>]]}, projection=[l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate], file_type=parquet, predicate=l_shipdate@10 <= 1998-09-02, pruning_predicate=l_shipdate_null_count@1 != row_count@2 AND l_shipdate_min@0 <= 1998-09-02, required_guarantees=[]
    └──────────────────────────────────────────────────

A real production service constructed on top of datafusion-distributed might want to log this plan at the end of each query so that developers can afterwards look for it and inspect it.

On huge scales, this plan is so big that it exceeds the limits of what's appropriate to put in a log, so depending on the logging infrastructure it can get truncated while incurring in a high storage cost.

This project is capable of representing DataFusion metrics in a protobuf format, and at the same time, all DataFusion plans can be serialized as well, so it should be possible for this project to provide a loggable compressed EXPLAIN ANALYZE-like representation that is suitable for a logging system + the tools for properly decoding and visualizing that.

The typical flow would look like this:

1. A distributed query is performed
2. Runtime metrics are collected at the coordinating stage
3. The DistributedExec node (or something else) has a method for extracting a compressed b64 string that represent the plan with metrics (or at least just the metrics)
4. The b64 string gets logged
5. This project's console CLI is capable of decoding that b64 string an provide nice tools for visualizing the plan

This opens the door to also collect several snapshots of runtime metrics at different timestamps of the query, and then provide a reconstruction of the runtime metrics at different moments in time for people to visualize how the query evolved.

[Rich-T-kid]

take