execution_plans: add MetricsCollectingStream

[jayshrivastava]

This change adds a new type called MetricsCollectingStream. It wraps a stream of FlightData
and collects any metrics that are passed in the app_metadata. This change also introduces
an FlightAppMetadata enum proto which can be used to define our app metadata protocol.

Informs #123

[NGA-TRAN]

The code looks good. I just have some minor suggestions for the tests

[NGA-TRAN]

Can you describe the purpose of this test? Do you want to test collecting metrics for one query with 2 stages or you want to test collecting metrics for 2 queries each has one stage?

If you want to test both, you may want to split them in 2 different tests and make it clear in the names of the tests or add comments to explain so. Also, do we collect metrics of different queries in one pass and need such tests?

Either way, you want to make the test valid by:

• Have the same query_id for the test on the same query. And have different query_id if they are for 2 different queries
• Use smaller number for child stage_id

[jayshrivastava]

Can you describe the purpose of this test?

I added a comment.

Do you want to test collecting metrics for one query with 2 stages or you want to test collecting metrics for 2 queries each has one stage?

This metrics stream doesn't really know the semantics of how queries are executed. The only "guarantee" it offers is that it collects metrics by StageKey but it doesn't know what those are. That's why I just used random StageKeys. However, I do agree that it's confusing, so I updated the test to use same query_id and stage_id with two different task ids. This should make more sense.

[NGA-TRAN]

This test is a bit long. What do you think to make different functions for creating metrics, stages and result verification, then create a macro to use them. There are many examples for that in Vanilla DF tests and I think Gabriel has that in this repro, too. That way, you can reuse them and it is easier for us to review

[NGA-TRAN]

@gabotechs : I am a bit ambiguous here. It would be great, if you can hep point Jayant to some good examples how to do so

[jayshrivastava]

Fixed. Plus I made it much more concise.

[jayshrivastava]

Added a test utils for metrics creation as well :) Now, you can call make_test_metrics_set_proto_from_seed to make test data. Also all metrics protos implement Eq now so it's easy to assert metrics equality in tests

[gabotechs]

Awesome! not much to say, this looks just right

[gabotechs]

I wonder if this should instead be:

#[pin_project]
pub struct MetricsCollectingStream<S>
where
    S: Stream<Item = Result<FlightData, FlightError>> + Send + Unpin,
{
    #[pin]
    inner: S,

I don't know the details about when this is needed, but I usually see wrapping streams be implemented this way in Rust.

[jayshrivastava]

👀 Will look into it.