Don't sort batches during plan

[tustvold]

Which issue does this PR close?

Closes #1939.

Rationale for this change

Currently SortExec performs the sort during the query plan

What changes are included in this PR?

This makes the SortExec lazy

Are there any user-facing changes?

SortExec no longer computes values during query planning

[tustvold]

A nicer fix would be to make ExternalSort stream, or less async, or some combination of the two, but in the short term this is a quick fix

[alamb]

Suggested change

	/// Combines a [`BoxStream`] with [`SchemaRef`] implementing
	/// [`RecordBatchStream`] for the combination
	/// Combines a [`BoxStream`] (created by calling `boxed` on a stream that produced `RecordBatch`es)
	/// with a [`SchemaRef`] implementing [`RecordBatchStream`] for the combination

[alamb]

I find the whole boxed stream concept somewhat confusing, so I am trying to leave hints around for my future self

[tustvold]

I've updated it to no longer need a boxed stream, the trade-off is more pinning nonsense 😅 PTAL

[alamb]

Looks better now (though I will be honest I don't really understand how pin_project works and how the code generated by this will be different than a Boxd stream 🤷

[alamb]

👍

[tustvold]

It'll have one-less stack allocation per stream, and one less dyn-dispatch per poll. Whether that matters 🤷‍♂️

[alamb]

It'll have one-less stack allocation per stream, and one less dyn-dispatch per poll. Whether that matters 🤷‍♂️

What I was getting at was that I don't know what sort of magic occurs as part of pin_project there may be more allocations I am not aware of. I am not worried, just opining

[tustvold]

It's really just a trick to ensure a given field is consistently either structurally or non-structurally pinned - https://doc.rust-lang.org/nightly/core/pin/index.html#projections-and-structural-pinning.

Unlike say async_trait, there isn't that much wizardry occuring under the hood, the generated code is actually relatively straightforward, well as straightforward as anything to do with pinning 😆

pub(crate) fn project<'__pin>(
    self: ::pin_project_lite::__private::Pin<&'__pin mut Self>,
) -> Projection<'__pin, S> {
    unsafe {
        let Self {
            schema, stream
        } = self.get_unchecked_mut();
        Projection {
            schema: schema,
            stream
            : ::pin_project_lite::__private::Pin::new_unchecked(stream
            ),
        }
    }
}

[alamb]

It's really just a trick to ensure a given field is consistently either s

TIL